hy-app-monitor/monitors/yumi/compose_templates.py

610 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

from __future__ import annotations
# compose 模板生成需要解析 docker inspect 的 JSON。
import json
# 本地模板目录和远端运行目录都走现有配置。
from pathlib import Path
# 远端命令和文件路径都需要安全转义。
import shlex
# 一些字段要做正则判断。
import re
from typing import Any
from core.config import DASHBOARD_CDC_SERVICE_NAME, HARBOR_PROJECT, HARBOR_REGISTRY, ROOT, RUNTIME_BASE_DIR, load_hosts, utc_now
from .runtime_config import read_remote_text, write_remote_text
from .ssh_remote import copy_remote_file, run_host_script_checked
# 本地完整 compose 模板统一放到 run 目录,避免把运行期产物写进 Git。
COMPOSE_TEMPLATE_DIR = ROOT / "run" / "compose_templates"
DEFAULT_JSON_LOG_OPTIONS = {"max-size": "100m", "max-file": "5"}
def managed_compose_hosts() -> list[dict[str, Any]]:
# 只维护当前 targets.json 里真正承载业务服务的宿主机。
return [host for host in load_hosts() if host.get("services")]
def managed_compose_host_names() -> list[str]:
# 页面和运维脚本都更适合拿主机名列表。
return [host["host"] for host in managed_compose_hosts()]
def managed_host_map() -> dict[str, dict[str, Any]]:
# 方便按主机名快速定位实例 ID 和服务清单。
return {host["host"]: host for host in managed_compose_hosts()}
def compose_template_path(host_name: str) -> Path:
# 每台机器一份本地模板文件。
return COMPOSE_TEMPLATE_DIR / f"{host_name}.yml"
def compose_remote_path(host_name: str) -> str:
# 远端 compose 路径沿用现网目录结构。
return f"{RUNTIME_BASE_DIR.rstrip('/')}/{host_name}/docker-compose.yml"
def container_name(host_name: str, service_name: str) -> str:
# 当前线上容器命名规则固定。
return f"likei-{host_name}-{service_name}"
def docker_inspect_payload(host: dict[str, Any], service_names: list[str]) -> list[dict[str, Any]]:
# 这里只抓 compose 需要的关键字段,避免远端 inspect 输出塞进一大坨无关 JSON。
container_names = [container_name(host["host"], service_name) for service_name in service_names]
if not container_names:
raise RuntimeError(f"no compose services configured for host: {host['host']}")
quoted_names = " ".join(shlex.quote(name) for name in container_names)
python_code = """
import json
import sys
payload = json.load(sys.stdin)
for item in payload:
config = item.get("Config") or {}
host_config = item.get("HostConfig") or {}
print(
json.dumps(
{
"Name": item.get("Name"),
"Config": {
"Image": config.get("Image"),
"StopSignal": config.get("StopSignal"),
"WorkingDir": config.get("WorkingDir"),
"Entrypoint": config.get("Entrypoint"),
"Cmd": config.get("Cmd"),
"Healthcheck": config.get("Healthcheck"),
},
"HostConfig": {
"NetworkMode": host_config.get("NetworkMode"),
"RestartPolicy": {"Name": (host_config.get("RestartPolicy") or {}).get("Name")},
"StopTimeout": host_config.get("StopTimeout"),
},
"Mounts": item.get("Mounts") or [],
},
ensure_ascii=False,
)
)
""".strip()
script = f"""set -eu
docker inspect {quoted_names} | python3 -c {shlex.quote(python_code)}
"""
result = run_host_script_checked(
host,
script,
command_name=f"inspect-compose-{host['host']}",
timeout_seconds=90,
)
payload: list[dict[str, Any]] = []
for raw_line in str(result.get("output") or "").splitlines():
line = raw_line.strip()
if not line:
continue
try:
payload.append(json.loads(line))
except json.JSONDecodeError as exc:
raise RuntimeError(f"invalid inspect field payload on {host['host']}") from exc
if not payload:
raise RuntimeError(f"empty docker inspect payload on {host['host']}")
return payload
def ns_to_duration_text(raw_value: int | None) -> str | None:
# docker inspect 的健康检查时间单位是纳秒compose 更适合秒级文本。
if not raw_value:
return None
if raw_value % 1_000_000_000 == 0:
return f"{raw_value // 1_000_000_000}s"
if raw_value % 1_000_000 == 0:
return f"{raw_value // 1_000_000}ms"
return str(raw_value)
def stop_grace_period_text(raw_value: int | None) -> str | None:
# docker inspect 的 StopTimeout 单位本来就是秒。
if raw_value is None or raw_value <= 0:
return None
return f"{int(raw_value)}s"
def image_is_pullable(image_ref: str) -> bool:
# 纯十六进制镜像 ID 不能当成远端可拉取镜像。
text = str(image_ref or "").strip()
if not text:
return False
return re.fullmatch(r"[0-9a-f]{12,64}", text) is None
def normalize_command(raw_value: Any) -> list[str] | str | None:
# docker inspect 的 Cmd / Entrypoint 可能是数组、空值或字符串。
if raw_value is None:
return None
if isinstance(raw_value, list):
return [str(item) for item in raw_value]
if isinstance(raw_value, str):
return raw_value
return [str(item) for item in raw_value]
def normalize_prod_profile_command(raw_value: list[str] | str | None) -> list[str] | str | None:
# 线上 compose 模板强制使用 prod避免从运行态刷新模板时把 local 启动参数带回来。
if raw_value is None:
return None
if isinstance(raw_value, list):
return [normalize_prod_profile_text(item) for item in raw_value]
return normalize_prod_profile_text(raw_value)
def normalize_prod_profile_text(raw_value: str) -> str:
text = str(raw_value)
for property_name in ("spring.profiles.active", "spring-profile.active"):
escaped = re.escape(property_name)
text = re.sub(rf"(--{escaped}=)[^\s]+", rf"\1prod", text)
text = re.sub(rf"(--{escaped})(\s+)[^\s]+", rf"\1\2prod", text)
return text
def healthcheck_block(config: dict[str, Any]) -> dict[str, Any] | None:
# 把 inspect 里的健康检查结构翻回 compose 可接受格式。
healthcheck = config.get("Healthcheck") or {}
test = healthcheck.get("Test")
if not test:
return None
block: dict[str, Any] = {"test": [str(item) for item in test]}
interval = ns_to_duration_text(int(healthcheck.get("Interval") or 0))
timeout = ns_to_duration_text(int(healthcheck.get("Timeout") or 0))
start_period = ns_to_duration_text(int(healthcheck.get("StartPeriod") or 0))
retries = int(healthcheck.get("Retries") or 0)
if interval:
block["interval"] = interval
if timeout:
block["timeout"] = timeout
if start_period:
block["start_period"] = start_period
if retries > 0:
block["retries"] = retries
return block
def render_compose_yaml(service_map: dict[str, dict[str, Any]]) -> str:
# 这里按需导入 PyYAML和现有项目依赖策略保持一致。
try:
import yaml
except Exception as exc: # noqa: BLE001
raise RuntimeError(f"yaml renderer unavailable: {exc}") from exc
payload = {"services": service_map}
return yaml.safe_dump(payload, allow_unicode=True, sort_keys=False).rstrip() + "\n"
def parse_compose_yaml(content: str) -> dict[str, Any]:
# 服务更新需要读取本地模板并直接改 image 字段。
try:
import yaml
except Exception as exc: # noqa: BLE001
raise RuntimeError(f"yaml parser unavailable: {exc}") from exc
payload = yaml.safe_load(content) or {}
if not isinstance(payload, dict):
raise RuntimeError("invalid compose yaml payload")
services = payload.get("services") or {}
if not isinstance(services, dict):
raise RuntimeError("invalid compose yaml services block")
return {"services": {str(name): dict(value or {}) for name, value in services.items()}}
def render_volume_specs(host_name: str, mounts: list[dict[str, Any]]) -> list[str]:
# 把 docker inspect 的挂载信息翻回 compose volumes。
host_dir = f"{RUNTIME_BASE_DIR.rstrip('/')}/{host_name}/"
specs: list[str] = []
for mount in mounts:
mount_type = str(mount.get("Type") or "").strip()
source = str(mount.get("Source") or "").strip()
destination = str(mount.get("Destination") or "").strip()
if mount_type != "bind" or not source or not destination:
continue
rendered_source = source
if source.startswith(host_dir):
rendered_source = f"./{source[len(host_dir):]}"
volume_spec = f"{rendered_source}:{destination}"
if not bool(mount.get("RW")):
volume_spec += ":ro"
specs.append(volume_spec)
return specs
def uses_local_runtime_artifact(spec: str) -> bool:
# 这些挂载代表服务依赖宿主机上的 jar / 二进制,不再适合镜像化发布。
text = str(spec or "").strip()
return (
text.endswith(":/application/service.jar:ro")
or text.endswith(":/application/service.jar")
or text.endswith(":/application/service:ro")
or text.endswith(":/application/service")
)
def strip_local_runtime_artifacts(volume_specs: list[str]) -> list[str]:
# 发布到 Harbor 后,需要把本地运行时制品挂载从模板里移掉。
return [spec for spec in volume_specs if not uses_local_runtime_artifact(spec)]
def build_service_definition(
host_name: str,
service_name: str,
inspect_payload: dict[str, Any],
*,
include_env_file: bool,
) -> dict[str, Any]:
# 只抽 compose 真正需要的字段,避免把 docker inspect 的噪声全部带回来。
config = inspect_payload.get("Config") or {}
host_config = inspect_payload.get("HostConfig") or {}
service_definition: dict[str, Any] = {}
image_ref = str(config.get("Image") or "").strip()
if not image_ref:
raise RuntimeError(f"missing image ref for {host_name}/{service_name}")
service_definition["image"] = image_ref
service_definition["container_name"] = str(inspect_payload.get("Name") or f"/{container_name(host_name, service_name)}").lstrip("/")
network_mode = str(host_config.get("NetworkMode") or "").strip()
if network_mode:
service_definition["network_mode"] = network_mode
restart_policy = str((host_config.get("RestartPolicy") or {}).get("Name") or "").strip()
if restart_policy:
service_definition["restart"] = restart_policy
log_config = host_config.get("LogConfig") or {}
log_driver = str(log_config.get("Type") or "").strip()
log_options = {
str(key): str(value)
for key, value in (log_config.get("Config") or {}).items()
if str(key).strip() and value is not None
}
if log_driver and log_driver != "json-file":
logging_definition: dict[str, Any] = {"driver": log_driver}
if log_options:
logging_definition["options"] = log_options
service_definition["logging"] = logging_definition
else:
# Keep json-file logs bounded even when Docker daemon defaults are not visible in inspect.
service_definition["logging"] = {
"driver": "json-file",
"options": {**log_options, **DEFAULT_JSON_LOG_OPTIONS},
}
stop_signal = str(config.get("StopSignal") or "").strip()
if stop_signal:
service_definition["stop_signal"] = stop_signal
stop_grace_period = stop_grace_period_text(host_config.get("StopTimeout"))
if stop_grace_period:
service_definition["stop_grace_period"] = stop_grace_period
if include_env_file:
service_definition["env_file"] = ["./service.env"]
volume_specs = render_volume_specs(host_name, list(inspect_payload.get("Mounts") or []))
if volume_specs:
service_definition["volumes"] = volume_specs
# 如果运行时依赖本地挂载的 jar / 二进制,就不要再强制拉镜像。
uses_local_runtime_artifact = any(
spec.endswith(":/application/service.jar:ro")
or spec.endswith(":/application/service")
or spec.endswith(":/application/service:ro")
for spec in volume_specs
)
if uses_local_runtime_artifact:
service_definition["pull_policy"] = "never"
elif image_is_pullable(image_ref):
service_definition["pull_policy"] = "always"
working_dir = str(config.get("WorkingDir") or "").strip()
if working_dir:
service_definition["working_dir"] = working_dir
entrypoint = normalize_command(config.get("Entrypoint"))
if entrypoint is not None:
service_definition["entrypoint"] = entrypoint
command = normalize_command(config.get("Cmd"))
if command is not None:
command = normalize_prod_profile_command(command)
service_definition["command"] = command
healthcheck = healthcheck_block(config)
if healthcheck:
service_definition["healthcheck"] = healthcheck
return service_definition
def build_dashboard_cdc_service_definition(host_name: str) -> dict[str, Any]:
# dashboard-cdc-worker 首次上线时还没有容器可 inspect只能从约定生成模板。
return {
"image": f"{HARBOR_REGISTRY}/{HARBOR_PROJECT}/dashboard-cdc-worker:bootstrap",
"container_name": f"likei-{host_name}-{DASHBOARD_CDC_SERVICE_NAME}",
"network_mode": "host",
"restart": "unless-stopped",
"stop_signal": "SIGTERM",
"env_file": ["./service.env", "./dashboard-cdc-worker.env"],
"pull_policy": "always",
"working_dir": "/application",
"entrypoint": [],
"command": ["/application/service"],
"healthcheck": {
"test": ["CMD-SHELL", "wget -qO- http://127.0.0.1:2910/health | grep -q '\"code\":\"ok\"'"],
"interval": "10s",
"timeout": "10s",
"retries": 30,
},
"logging": {
"driver": "json-file",
"options": {"max-size": "200m", "max-file": "5"},
},
}
def generate_host_compose_template(host_name: str) -> dict[str, Any]:
# 生成模板时严格按 targets.json 里的服务顺序输出,后续 diff 更稳定。
host_map = managed_host_map()
host = host_map.get(host_name)
if host is None:
raise RuntimeError(f"unsupported compose host: {host_name}")
service_names = [str(service.get("name") or "").strip() for service in host.get("services") or []]
service_names = [service_name for service_name in service_names if service_name]
if not service_names:
raise RuntimeError(f"host has no managed services: {host_name}")
inspect_items = docker_inspect_payload(host, service_names)
inspect_by_name = {
str((item.get("Name") or "").lstrip("/")): item
for item in inspect_items
}
service_map: dict[str, dict[str, Any]] = {}
missing_services: list[str] = []
for service_name in service_names:
expected_container = container_name(host_name, service_name)
inspect_item = inspect_by_name.get(expected_container)
if inspect_item is None:
if service_name == DASHBOARD_CDC_SERVICE_NAME:
service_map[service_name] = build_dashboard_cdc_service_definition(host_name)
continue
missing_services.append(service_name)
continue
service_map[service_name] = build_service_definition(
host_name,
service_name,
inspect_item,
include_env_file=True,
)
if missing_services:
raise RuntimeError(f"missing running containers on {host_name}: {', '.join(missing_services)}")
content = render_compose_yaml(service_map)
return {
"ok": True,
"host": host_name,
"services": list(service_map.keys()),
"updatedAt": utc_now(),
"content": content,
"path": str(compose_template_path(host_name)),
}
def save_local_host_compose_template(host_name: str) -> dict[str, Any]:
# 从远端运行态重新生成本地完整模板。
payload = generate_host_compose_template(host_name)
path = compose_template_path(host_name)
path.parent.mkdir(parents=True, exist_ok=True)
previous = path.read_text(encoding="utf-8") if path.exists() else ""
path.write_text(payload["content"], encoding="utf-8")
return {
**payload,
"changed": previous != payload["content"],
"source": "remote-runtime",
}
def load_local_host_compose_template(host_name: str) -> dict[str, Any]:
# 本地模板优先作为远端同步的稳定来源。
path = compose_template_path(host_name)
if not path.exists():
raise RuntimeError(f"local compose template missing: {host_name}")
return {
"ok": True,
"host": host_name,
"services": [str(service.get("name") or "").strip() for service in managed_host_map()[host_name].get("services") or []],
"updatedAt": utc_now(),
"content": path.read_text(encoding="utf-8"),
"path": str(path),
"source": "local-cache",
}
def ensure_local_host_compose_template(host_name: str) -> dict[str, Any]:
# 第一次进入服务更新时,模板可能还没在本地生成。
path = compose_template_path(host_name)
if path.exists():
return load_local_host_compose_template(host_name)
return save_local_host_compose_template(host_name)
def load_local_host_compose_model(host_name: str) -> dict[str, Any]:
# 同时返回解析后的 services方便直接修改模板。
template_payload = ensure_local_host_compose_template(host_name)
parsed = parse_compose_yaml(str(template_payload.get("content") or ""))
return {
**template_payload,
"services": parsed["services"],
}
def save_local_host_compose_model(host_name: str, service_map: dict[str, dict[str, Any]]) -> dict[str, Any]:
# 模板编辑完成后,统一走这里回写磁盘并返回 diff 信息。
path = compose_template_path(host_name)
path.parent.mkdir(parents=True, exist_ok=True)
content = render_compose_yaml(service_map)
previous = path.read_text(encoding="utf-8") if path.exists() else ""
path.write_text(content, encoding="utf-8")
return {
"ok": True,
"host": host_name,
"updatedAt": utc_now(),
"path": str(path),
"content": content,
"changed": previous != content,
"source": "local-cache",
"services": list(service_map.keys()),
}
def update_local_service_image(host_name: str, service_name: str, image_ref: str) -> dict[str, Any]:
# 服务更新链路会把新镜像 digest 写回本地完整模板。
if not str(image_ref or "").strip():
raise RuntimeError("image ref is required")
model = load_local_host_compose_model(host_name)
service_map = dict(model.get("services") or {})
service_definition = dict(service_map.get(service_name) or {})
if not service_definition:
raise RuntimeError(f"missing service in local compose template: {host_name}/{service_name}")
service_definition["image"] = str(image_ref).strip()
volume_specs = [str(item) for item in service_definition.get("volumes") or []]
volume_specs = strip_local_runtime_artifacts(volume_specs)
if volume_specs:
service_definition["volumes"] = volume_specs
else:
service_definition.pop("volumes", None)
if image_is_pullable(image_ref):
service_definition["pull_policy"] = "always"
else:
service_definition.pop("pull_policy", None)
service_map[service_name] = service_definition
saved = save_local_host_compose_model(host_name, service_map)
saved["service"] = service_name
saved["image"] = str(image_ref).strip()
return saved
def verify_remote_compose(host: dict[str, Any], remote_path: str) -> list[str]:
# 下发后立即让 docker compose 解析一遍,避免把坏文件写到线上却没人发现。
script = f"""set -eu
FILE={shlex.quote(remote_path)}
docker compose -f "$FILE" config --services
"""
result = run_host_script_checked(
host,
script,
command_name=f"verify-compose-{host['host']}",
timeout_seconds=60,
)
return [line.strip() for line in str(result.get("output") or "").splitlines() if line.strip()]
def sync_remote_host_compose_template(host_name: str, *, refresh_from_remote: bool = True) -> dict[str, Any]:
# 对外暴露的主入口:优先刷新本地模板,再把完整 compose 同步到远端。
host_map = managed_host_map()
host = host_map.get(host_name)
if host is None:
raise RuntimeError(f"unsupported compose host: {host_name}")
template_payload: dict[str, Any] | None = None
refresh_error = ""
if refresh_from_remote:
try:
template_payload = save_local_host_compose_template(host_name)
except Exception as exc: # noqa: BLE001
refresh_error = str(exc)
if template_payload is None:
template_payload = load_local_host_compose_template(host_name)
template_payload["refreshError"] = refresh_error
remote_path = compose_remote_path(host_name)
try:
remote_current = read_remote_text(host, remote_path)
except Exception:
remote_current = ""
backup_path = ""
if remote_current and remote_current != template_payload["content"]:
backup_path = f"{remote_path}.hyapp.{utc_now().replace(':', '').replace('+', '_')}.bak"
copy_remote_file(host, remote_path, backup_path)
changed = remote_current != template_payload["content"]
if changed:
write_remote_text(host, remote_path, template_payload["content"])
remote_after = read_remote_text(host, remote_path)
if remote_after != template_payload["content"]:
raise RuntimeError(f"compose verify content mismatch on {host_name}")
remote_services = verify_remote_compose(host, remote_path)
return {
"ok": True,
"host": host_name,
"services": template_payload["services"],
"updatedAt": utc_now(),
"path": remote_path,
"templatePath": template_payload["path"],
"changed": changed,
"backupPath": backup_path,
"remoteServices": remote_services,
"source": template_payload.get("source") or "local-cache",
"refreshError": template_payload.get("refreshError") or "",
}
def sync_remote_compose_templates(host_names: list[str], *, refresh_from_remote: bool = True) -> dict[str, Any]:
# 允许对一组主机做完整模板同步。
normalized_names = [str(name or "").strip() for name in host_names if str(name or "").strip()]
normalized_names = list(dict.fromkeys(normalized_names))
if not normalized_names:
normalized_names = managed_compose_host_names()
items = [sync_remote_host_compose_template(host_name, refresh_from_remote=refresh_from_remote) for host_name in normalized_names]
return {
"ok": True,
"updatedAt": utc_now(),
"hosts": items,
}