#!/usr/bin/env python3 from __future__ import annotations import argparse import base64 import json import os import shlex import sys import time from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from typing import Any def load_env_file(path: str) -> None: # hy-app-monitor 现有 .env 可直接复用；这里只注入缺失变量，不覆盖调用方显式 export 的值。 if not path: return env_path = Path(path).expanduser().resolve() if not env_path.exists(): raise RuntimeError(f"env file not found: {env_path}") for raw_line in env_path.read_text(encoding="utf-8", errors="replace").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) key = key.strip() value = value.strip().strip('"').strip("'") if key and key not in os.environ: os.environ[key] = value def load_inventory(path: str) -> dict[str, Any]: inventory_path = Path(path).expanduser().resolve() with inventory_path.open("r", encoding="utf-8") as handle: inventory = json.load(handle) validate_inventory(inventory) return inventory def validate_inventory(inventory: dict[str, Any]) -> None: # 发布脚本只认明确的 host/service 映射，避免误操作同机已有 Java 项目。 if not isinstance(inventory.get("hosts"), dict) or not inventory["hosts"]: raise RuntimeError("inventory.hosts is required") if not isinstance(inventory.get("services"), dict) or not inventory["services"]: raise RuntimeError("inventory.services is required") for host_name, host in inventory["hosts"].items(): if not str(host.get("instance_id") or "").strip(): raise RuntimeError(f"hosts.{host_name}.instance_id is required") if not str(host.get("private_ip") or "").strip(): raise RuntimeError(f"hosts.{host_name}.private_ip is required") for service_name, service in inventory["services"].items(): for key in ("unit", "container", "env_file", "target_port", "hosts"): if key not in service: raise RuntimeError(f"services.{service_name}.{key} is required") for host_name in service["hosts"]: if host_name not in inventory["hosts"]: raise RuntimeError(f"services.{service_name}.hosts contains unknown host: {host_name}") def csv_values(text: str) -> list[str]: return [item.strip() for item in str(text or "").split(",") if item.strip()] def selected_service_names(inventory: dict[str, Any], raw_services: str) -> list[str]: names = csv_values(raw_services) if not names: raise RuntimeError("--services is required") unknown = [name for name in names if name not in inventory["services"]] if unknown: raise RuntimeError(f"unsupported services: {', '.join(unknown)}") return list(dict.fromkeys(names)) def selected_host_names(service: dict[str, Any], raw_hosts: str) -> list[str]: hosts = list(dict.fromkeys(str(item) for item in service.get("hosts") or [])) requested = csv_values(raw_hosts) if not requested: return hosts missing = [host for host in requested if host not in hosts] if missing: raise RuntimeError(f"selected hosts are not assigned to this service: {', '.join(missing)}") return requested def image_overrides(items: list[str]) -> dict[str, str]: overrides: dict[str, str] = {} for item in items: if "=" not in item: raise RuntimeError("--image must use service=image") service_name, image = item.split("=", 1) service_name = service_name.strip() image = image.strip() if not service_name or not image: raise RuntimeError("--image must use non-empty service=image") overrides[service_name] = image return overrides def render_image(inventory: dict[str, Any], service_name: str, tag: str, overrides: dict[str, str]) -> str: if service_name in overrides: return overrides[service_name] if not tag: raise RuntimeError(f"--tag or --image {service_name}=... is required for deploy") service = inventory["services"][service_name] registry = str(inventory.get("registry") or "").rstrip("/") template = str(service.get("image_template") or "${REGISTRY}/" + service_name + ":${TAG}") return template.replace("${REGISTRY}", registry).replace("${TAG}", tag) def build_plan( inventory: dict[str, Any], *, action: str, service_names: list[str], raw_hosts: str, tag: str, overrides: dict[str, str], no_clb: bool, ) -> list[dict[str, Any]]: # 顺序是服务优先、实例串行；多服务发布不会同时摘掉同一组双机实例。 plan: list[dict[str, Any]] = [] for service_name in service_names: service = inventory["services"][service_name] image = render_image(inventory, service_name, tag, overrides) if action == "deploy" else "" for host_name in selected_host_names(service, raw_hosts): host = inventory["hosts"][host_name] clb = service.get("clb") or {} plan.append( { "action": action, "service": service_name, "host": host_name, "instance_id": host["instance_id"], "private_ip": host["private_ip"], "target_port": int(service["target_port"]), "unit": service["unit"], "container": service["container"], "env_file": service["env_file"], "config_path": service.get("config_path") or f"/etc/hyapp/{service_name}/config.yaml", "image": image, "clb_enabled": bool(clb.get("enabled")) and not no_clb, "clb_load_balancer_id": str(clb.get("load_balancer_id") or ""), "clb_listener_ids": list(clb.get("listener_ids") or []), "drain_seconds": int(clb.get("drain_seconds") or 0), } ) return plan def secret_id() -> str: return os.environ.get("TENCENTCLOUD_SECRET_ID") or os.environ.get("TENCENT_SECRET_ID") or "" def secret_key() -> str: return os.environ.get("TENCENTCLOUD_SECRET_KEY") or os.environ.get("TENCENT_SECRET_KEY") or "" def region(inventory: dict[str, Any]) -> str: return os.environ.get("TENCENTCLOUD_REGION") or os.environ.get("DEPLOY_REGION") or str(inventory.get("region") or "") def require_tencent_credentials(inventory: dict[str, Any]) -> tuple[str, str, str]: sid = secret_id() skey = secret_key() reg = region(inventory) missing = [] if not sid: missing.append("TENCENTCLOUD_SECRET_ID") if not skey: missing.append("TENCENTCLOUD_SECRET_KEY") if not reg: missing.append("TENCENTCLOUD_REGION") if missing: raise RuntimeError("missing Tencent Cloud credentials: " + ", ".join(missing)) return sid, skey, reg def build_tat_client(inventory: dict[str, Any]) -> Any: sid, skey, reg = require_tencent_credentials(inventory) try: from tencentcloud.common import credential from tencentcloud.common.profile.client_profile import ClientProfile from tencentcloud.common.profile.http_profile import HttpProfile from tencentcloud.tat.v20201028 import tat_client except Exception as exc: # noqa: BLE001 raise RuntimeError(f"tencentcloud sdk unavailable: {exc}") from exc return tat_client.TatClient( credential.Credential(sid, skey), reg, ClientProfile(httpProfile=HttpProfile(endpoint="tat.tencentcloudapi.com")), ) def build_clb_client(inventory: dict[str, Any]) -> Any: sid, skey, reg = require_tencent_credentials(inventory) try: from tencentcloud.clb.v20180317 import clb_client from tencentcloud.common import credential from tencentcloud.common.profile.client_profile import ClientProfile from tencentcloud.common.profile.http_profile import HttpProfile except Exception as exc: # noqa: BLE001 raise RuntimeError(f"tencentcloud sdk unavailable: {exc}") from exc return clb_client.ClbClient( credential.Credential(sid, skey), reg, ClientProfile(httpProfile=HttpProfile(endpoint="clb.tencentcloudapi.com")), ) def invoke_tencent(action: Any, label: str, retries: int = 3) -> Any: try: from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException except Exception: # noqa: BLE001 TencentCloudSDKException = Exception # type: ignore[assignment] last_error: Exception | None = None for attempt in range(1, retries + 1): try: return action() except TencentCloudSDKException as exc: # type: ignore[misc] last_error = exc if attempt == retries: break time.sleep(2 * attempt) raise RuntimeError(f"{label} failed: {last_error}") from last_error def decode_tat_output(output: str) -> str: if not output: return "" return base64.b64decode(output).decode("utf-8", errors="replace") def run_tat_shell( client: Any, *, instance_id: str, script: str, command_name: str, timeout_seconds: int, poll_seconds: int, ) -> dict[str, Any]: # TAT 是本方案唯一远端入口；脚本不会要求目标机开放 SSH。 from tencentcloud.tat.v20201028 import models as tat_models request = tat_models.RunCommandRequest() request.from_json_string( json.dumps( { "CommandName": command_name, "CommandType": "SHELL", "Content": base64.b64encode(script.encode("utf-8")).decode("ascii"), "InstanceIds": [instance_id], "Timeout": timeout_seconds, "WorkingDirectory": "/root", } ) ) response = json.loads(invoke_tencent(lambda: client.RunCommand(request), "RunCommand").to_json_string()) invocation_id = response["InvocationId"] deadline = time.time() + timeout_seconds + 60 while time.time() < deadline: time.sleep(max(poll_seconds, 1)) query = tat_models.DescribeInvocationTasksRequest() query.from_json_string( json.dumps( { "HideOutput": False, "Limit": 50, "Filters": [{"Name": "invocation-id", "Values": [invocation_id]}], } ) ) payload = json.loads(invoke_tencent(lambda: client.DescribeInvocationTasks(query), "DescribeInvocationTasks").to_json_string()) for task in payload.get("InvocationTaskSet") or []: if str(task.get("InstanceId") or "") != instance_id: continue status = str(task.get("TaskStatus") or "") if status in {"SUCCESS", "FAILED", "TIMEOUT", "CANCELLED", "START_FAILED"}: return { "status": status, "output": decode_tat_output(str((task.get("TaskResult") or {}).get("Output") or "")), } return {"status": "TIMEOUT", "output": ""} def call_clb(client: Any, method_name: str, payload: dict[str, Any]) -> dict[str, Any]: from tencentcloud.clb.v20180317 import models as clb_models request_class = getattr(clb_models, f"{method_name}Request") request = request_class() request.from_json_string(json.dumps(payload)) response = invoke_tencent(lambda: getattr(client, method_name)(request), method_name) decoded = json.loads(response.to_json_string()) if isinstance(decoded.get("Response"), dict): return dict(decoded["Response"]) return dict(decoded) def wait_clb_task(client: Any, task_id: str, *, timeout_seconds: int, poll_seconds: int) -> None: # ModifyTargetWeight 是异步任务，必须轮询到成功后才能重启实例。 deadline = time.time() + max(timeout_seconds, 1) while time.time() < deadline: payload = call_clb(client, "DescribeTaskStatus", {"TaskId": task_id}) status = int(payload.get("Status") if payload.get("Status") is not None else 2) if status == 0: return if status == 1: raise RuntimeError(f"CLB task failed: {payload.get('Message') or task_id}") time.sleep(max(poll_seconds, 1)) raise RuntimeError(f"CLB task timed out: {task_id}") def target_matches(step: dict[str, Any], target: dict[str, Any]) -> bool: instance_id = str(target.get("InstanceId") or target.get("TargetId") or "").strip() if instance_id and instance_id != step["instance_id"]: return False target_port = int(target.get("Port") or 0) if target_port and target_port != int(step["target_port"]): return False ips = [str(item or "").strip() for item in (target.get("PrivateIpAddresses") or [target.get("IP")]) if str(item or "").strip()] if ips and step["private_ip"] not in ips: return False return bool(instance_id or ips) def binding_key(binding: dict[str, Any]) -> str: return "|".join( [ str(binding.get("listener_id") or ""), str(binding.get("location_id") or ""), str(binding.get("instance_id") or ""), str(int(binding.get("port") or 0)), ] ) def collect_bindings(step: dict[str, Any], listeners: list[dict[str, Any]]) -> list[dict[str, Any]]: bindings: dict[str, dict[str, Any]] = {} for listener in listeners: listener_id = str(listener.get("ListenerId") or "").strip() for target in listener.get("Targets") or []: if target_matches(step, target): binding = make_binding(step, listener_id, "", target) bindings[binding_key(binding)] = binding for rule in listener.get("Rules") or []: location_id = str(rule.get("LocationId") or "").strip() for target in rule.get("Targets") or []: if target_matches(step, target): binding = make_binding(step, listener_id, location_id, target) bindings[binding_key(binding)] = binding return list(bindings.values()) def make_binding(step: dict[str, Any], listener_id: str, location_id: str, target: dict[str, Any]) -> dict[str, Any]: return { "listener_id": listener_id, "location_id": location_id, "instance_id": str(target.get("InstanceId") or target.get("TargetId") or step["instance_id"]).strip(), "port": int(target.get("Port") or step["target_port"]), "weight": int(target.get("Weight") or 0), } def prepare_clb_snapshot(client: Any, step: dict[str, Any]) -> dict[str, Any]: if should_auto_discover_clb(step): return discover_clb_snapshot(client, step) payload = call_clb( client, "DescribeTargets", { "LoadBalancerId": step["clb_load_balancer_id"], "ListenerIds": step["clb_listener_ids"], }, ) bindings = collect_bindings(step, list(payload.get("Listeners") or [])) if not bindings: return discover_clb_snapshot(client, step) return {"step": step, "bindings": bindings} def should_auto_discover_clb(step: dict[str, Any]) -> bool: load_balancer_id = str(step.get("clb_load_balancer_id") or "").strip() listener_ids = [str(item or "").strip() for item in step.get("clb_listener_ids") or []] if not load_balancer_id or "REPLACE" in load_balancer_id or load_balancer_id.upper() == "AUTO": return True return any((not item) or "REPLACE" in item or item.upper() == "AUTO" for item in listener_ids) def discover_load_balancers_for_backend(client: Any, step: dict[str, Any]) -> list[dict[str, Any]]: # 腾讯云 CLB 支持按后端私网 IP 反查负载均衡，避免手填内网 CLB ID。 offset = 0 found: list[dict[str, Any]] = [] while True: payload = call_clb( client, "DescribeLoadBalancers", { "BackendPrivateIps": [step["private_ip"]], "WithRs": 1, "Limit": 100, "Offset": offset, }, ) items = list(payload.get("LoadBalancerSet") or []) found.extend(items) total = int(payload.get("TotalCount") or len(found)) if len(found) >= total or not items: break offset += len(items) return found def discover_clb_snapshot(client: Any, step: dict[str, Any]) -> dict[str, Any]: load_balancers = discover_load_balancers_for_backend(client, step) matched: list[dict[str, Any]] = [] for load_balancer in load_balancers: load_balancer_id = str(load_balancer.get("LoadBalancerId") or "").strip() if not load_balancer_id: continue payload = call_clb(client, "DescribeTargets", {"LoadBalancerId": load_balancer_id}) bindings = collect_bindings(step, list(payload.get("Listeners") or [])) if bindings: matched.append({"load_balancer_id": load_balancer_id, "bindings": bindings}) if not matched: raise RuntimeError(f"CLB target not found for {step['service']} on {step['host']} {step['private_ip']}:{step['target_port']}") if len(matched) > 1: summary = ", ".join(item["load_balancer_id"] for item in matched) raise RuntimeError(f"multiple CLB bindings discovered for {step['service']} on {step['host']}: {summary}") discovered = matched[0] step["clb_load_balancer_id"] = discovered["load_balancer_id"] step["clb_listener_ids"] = sorted({binding["listener_id"] for binding in discovered["bindings"] if binding["listener_id"]}) return {"step": step, "bindings": discovered["bindings"], "discovered": True} def modify_binding_weight(client: Any, inventory: dict[str, Any], binding: dict[str, Any], weight: int, load_balancer_id: str) -> None: payload: dict[str, Any] = { "LoadBalancerId": load_balancer_id, "ListenerId": binding["listener_id"], "Targets": [{"InstanceId": binding["instance_id"], "Port": int(binding["port"])}], "Weight": int(weight), } if binding.get("location_id"): payload["LocationId"] = binding["location_id"] response = call_clb(client, "ModifyTargetWeight", payload) wait_clb_task( client, str(response.get("RequestId") or "").strip(), timeout_seconds=int(inventory.get("clb_task_timeout_seconds") or 180), poll_seconds=int(inventory.get("clb_poll_seconds") or 2), ) def set_clb_snapshot_weight(client: Any, inventory: dict[str, Any], snapshot: dict[str, Any], weight: int | None) -> None: step = snapshot["step"] for binding in snapshot["bindings"]: target_weight = int(binding["weight"]) if weight is None else int(weight) modify_binding_weight(client, inventory, binding, target_weight, step["clb_load_balancer_id"]) def remote_restart_script(service: dict[str, Any], *, image: str, health_timeout_seconds: int) -> str: # 远端只改当前 Go 服务的 docker.env 和 systemd unit，避免碰同机 Java compose。 lines = [ "set -euo pipefail", f"UNIT={shlex.quote(str(service['unit']))}", f"CONTAINER={shlex.quote(str(service['container']))}", f"ENV_FILE={shlex.quote(str(service['env_file']))}", f"HEALTH_TIMEOUT={int(health_timeout_seconds)}", ] if image: lines.extend( [ f"IMAGE={shlex.quote(image)}", 'test -f "$ENV_FILE"', 'TMP_FILE="$(mktemp)"', 'awk -v image="$IMAGE" \'BEGIN{done=0} /^IMAGE=/{print "IMAGE=" image; done=1; next} {print} END{if(!done) print "IMAGE=" image}\' "$ENV_FILE" > "$TMP_FILE"', 'install -m 0640 "$TMP_FILE" "$ENV_FILE"', 'rm -f "$TMP_FILE"', 'docker pull "$IMAGE"', ] ) lines.extend( [ 'systemctl restart "$UNIT"', 'deadline=$((SECONDS + HEALTH_TIMEOUT))', 'while ! systemctl is-active --quiet "$UNIT"; do', ' if [ "$SECONDS" -ge "$deadline" ]; then', ' systemctl status "$UNIT" --no-pager || true', ' exit 20', " fi", " sleep 2", "done", 'while [ "$SECONDS" -lt "$deadline" ]; do', ' state="$(docker inspect --format \'{{if .State.Health}}{{.State.Health.Status}}{{else}}{{.State.Status}}{{end}}\' "$CONTAINER" 2>/dev/null || true)"', ' if [ "$state" = "healthy" ] || [ "$state" = "running" ]; then', ' docker ps --filter "name=$CONTAINER" --format "{{.Names}} {{.Status}}"', " exit 0", " fi", " sleep 2", "done", 'docker inspect "$CONTAINER" 2>/dev/null || true', 'journalctl -u "$UNIT" -n 80 --no-pager || true', "exit 21", ] ) return "\n".join(lines) + "\n" def remote_discover_script(*, host_services: list[dict[str, Any]], include_config: bool, max_config_bytes: int) -> str: # 只读取 inventory 明确声明的 hyapp unit/env/container/config，不扫描或修改 Java 项目。 payload = { "includeConfig": include_config, "maxConfigBytes": max_config_bytes, "services": [ { "configPath": service.get("config_path") or "", "container": service.get("container") or "", "envFile": service.get("env_file") or "", "name": service.get("service") or "", "targetPort": service.get("target_port") or 0, "unit": service.get("unit") or "", } for service in host_services ], } blob = base64.b64encode(json.dumps(payload).encode("utf-8")).decode("ascii") script = r"""set -euo pipefail python3 - <<'PY' import base64 import datetime import json import os import re import subprocess payload = json.loads(base64.b64decode(__PAYLOAD__).decode("utf-8")) services = payload.get("services") or [] include_config = bool(payload.get("includeConfig")) max_config_bytes = int(payload.get("maxConfigBytes") or 12000) secret_key_re = re.compile(r"(secret|password|passwd|token|credential|private[_-]?key|access[_-]?key|sign[_-]?key|dsn)", re.I) def command(args, timeout=8): try: return subprocess.check_output(args, text=True, stderr=subprocess.STDOUT, timeout=timeout).strip() except Exception as exc: return str(exc).strip() def percent_value(value): match = re.search(r"[-+]?\d+(?:\.\d+)?", str(value or "")) return float(match.group(0)) if match else None def read_file(path, max_bytes): item = { "content": "", "exists": False, "path": path, "truncated": False, } if not path or not os.path.exists(path): return item item["exists"] = True with open(path, "r", encoding="utf-8", errors="replace") as handle: content = handle.read(max_bytes + 1) item["truncated"] = len(content) > max_bytes item["content"] = content[:max_bytes] return item def redact_text(text): redacted = [] for line in text.splitlines(): match = re.match(r"^(\s*[-\w.]+\s*(?::|=)\s*)(.*)$", line) if match and secret_key_re.search(match.group(1)): redacted.append(match.group(1) + '"***"') else: redacted.append(line) return "\n".join(redacted) + ("\n" if text.endswith("\n") else "") def parse_env(text): env = {} for raw_line in text.splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) key = key.strip() value = value.strip().strip('"').strip("'") env[key] = "***" if secret_key_re.search(key) else value return env def parse_docker_inspect(container): if not container: return {} raw = command(["docker", "inspect", container], timeout=10) try: decoded = json.loads(raw) except Exception: return {"inspectError": raw[-1000:]} return decoded[0] if decoded else {} def parse_docker_stats(container): if not container: return {} raw = command(["docker", "stats", "--no-stream", "--format", "{{json .}}", container], timeout=12) line = raw.splitlines()[0] if raw else "" try: stats = json.loads(line) except Exception: return {"statsError": raw[-1000:]} return { "cpuPercent": percent_value(stats.get("CPUPerc")), "memoryPercent": percent_value(stats.get("MemPerc")), "memoryUsage": stats.get("MemUsage") or "", } def parse_time(value): text = str(value or "") if not text or text.startswith("0001-"): return None try: return datetime.datetime.fromisoformat(text.replace("Z", "+00:00")) except Exception: return None def duration_text(value): started_at = parse_time(value) if started_at is None: return "未采集" now = datetime.datetime.now(datetime.timezone.utc) delta = now - started_at.astimezone(datetime.timezone.utc) seconds = max(int(delta.total_seconds()), 0) days, remainder = divmod(seconds, 86400) hours, remainder = divmod(remainder, 3600) minutes = remainder // 60 if days > 0: return f"{days} 天 {hours} 小时" if hours > 0: return f"{hours} 小时 {minutes} 分钟" return f"{minutes} 分钟" def image_tag(image): text = str(image or "") if ":" not in text: return text return text.rsplit(":", 1)[1] def service_status(unit_state, container_status, health_status): if unit_state == "active" and container_status == "running" and health_status in ("", "healthy"): return "healthy" if unit_state == "activating" or container_status in ("created", "restarting") or health_status == "starting": return "running" if unit_state == "active" and container_status == "running": return "warning" if unit_state or container_status or health_status: return "critical" return "unknown" items = [] for service in services: name = str(service.get("name") or "") unit = str(service.get("unit") or "") container = str(service.get("container") or "") env_file = str(service.get("envFile") or "") env_payload = read_file(env_file, max_config_bytes) env = parse_env(env_payload["content"]) config_path = env.get("CONFIG_PATH") or str(service.get("configPath") or "") or f"/etc/hyapp/{name}/config.yaml" config_payload = read_file(config_path, max_config_bytes) if include_config else { "content": "", "exists": os.path.exists(config_path) if config_path else False, "path": config_path, "truncated": False, } inspect = parse_docker_inspect(container) state = inspect.get("State") or {} config = inspect.get("Config") or {} health = (state.get("Health") or {}).get("Status") or "" image = env.get("IMAGE") or config.get("Image") or "" started_at = state.get("StartedAt") or "" created_at = inspect.get("Created") or "" stats = parse_docker_stats(container) target_port = int(service.get("targetPort") or 0) listeners = command(["bash", "-lc", f"ss -lntp 2>/dev/null | grep ':{target_port} ' || true"]) if target_port else "" logs = command(["journalctl", "-u", unit, "-n", "30", "--no-pager", "-o", "short-iso"], timeout=10) if include_config and unit else "" items.append({ "configExists": bool(config_payload["exists"]), "configPath": config_path, "configTruncated": bool(config_payload["truncated"]), "configYaml": redact_text(config_payload["content"]) if include_config and config_payload["exists"] else "", "container": container, "containerId": str(inspect.get("Id") or "")[:12], "containerStatus": str(state.get("Status") or ""), "cpuPercent": stats.get("cpuPercent"), "createdAt": created_at, "env": env, "envFile": env_file, "envFileExists": bool(env_payload["exists"]), "healthStatus": health, "image": image, "imageTag": image_tag(image), "listeners": listeners, "logs": logs, "memoryPercent": stats.get("memoryPercent"), "memoryUsage": stats.get("memoryUsage") or "", "name": name, "restartCount": int(inspect.get("RestartCount") or 0) if inspect else 0, "runningTime": duration_text(started_at), "startedAt": started_at, "status": service_status(command(["systemctl", "is-active", unit]) if unit else "", str(state.get("Status") or ""), health), "targetPort": target_port, "unit": unit, "unitState": command(["systemctl", "is-active", unit]) if unit else "", }) print(json.dumps({ "hostname": command(["hostname"]), "ok": True, "services": items, }, ensure_ascii=False)) PY """ return script.replace("__PAYLOAD__", repr(blob)) def execute_discovery( inventory: dict[str, Any], plan: list[dict[str, Any]], *, dry_run: bool, include_config: bool, max_config_bytes: int, ) -> dict[str, Any]: if dry_run: return {"ok": True, "dry_run": True, "steps": plan} clb_client = build_clb_client(inventory) seen_hosts = list(dict.fromkeys(step["host"] for step in plan)) by_host = {name: next(step for step in plan if step["host"] == name) for name in seen_hosts} services_by_host = {name: [step for step in plan if step["host"] == name] for name in seen_hosts} def discover_host(host_name: str, seed: dict[str, Any]) -> dict[str, Any]: try: tat_client = build_tat_client(inventory) result = run_tat_shell( tat_client, instance_id=seed["instance_id"], script=remote_discover_script( host_services=services_by_host[host_name], include_config=include_config, max_config_bytes=max_config_bytes, ), command_name=f"hyapp-discover-{host_name}", timeout_seconds=int(inventory.get("tat_timeout_seconds") or 900), poll_seconds=int(inventory.get("tat_poll_seconds") or 2), ) output = str(result.get("output") or "") try: payload = json.loads(output or "{}") except json.JSONDecodeError: payload = {} services = list(payload.get("services") or []) return { "host": host_name, "instance_id": seed["instance_id"], "private_ip": seed["private_ip"], "status": result["status"], "output": "" if services else output[-5000:], "services": services, } except Exception as exc: # noqa: BLE001 return { "host": host_name, "instance_id": seed["instance_id"], "private_ip": seed["private_ip"], "status": "ERROR", "error": str(exc), "services": [], } host_results: dict[str, dict[str, Any]] = {} max_workers = max(1, min(len(by_host), int(inventory.get("discover_parallelism") or 6))) with ThreadPoolExecutor(max_workers=max_workers) as executor: future_to_host = {executor.submit(discover_host, host_name, seed): host_name for host_name, seed in by_host.items()} for future in as_completed(future_to_host): host_name = future_to_host[future] host_results[host_name] = future.result() hosts = [host_results[name] for name in seen_hosts] clb_bindings: list[dict[str, Any]] = [] for step in plan: if not step["clb_enabled"]: continue try: snapshot = prepare_clb_snapshot(clb_client, dict(step)) clb_bindings.append( { "service": step["service"], "host": step["host"], "load_balancer_id": snapshot["step"]["clb_load_balancer_id"], "listener_ids": snapshot["step"]["clb_listener_ids"], "bindings": snapshot["bindings"], "discovered": bool(snapshot.get("discovered")), } ) except Exception as exc: # noqa: BLE001 clb_bindings.append( { "service": step["service"], "host": step["host"], "error": str(exc), } ) return {"ok": True, "hosts": hosts, "clb": clb_bindings} def execute_plan(inventory: dict[str, Any], plan: list[dict[str, Any]], *, dry_run: bool, assume_yes: bool) -> dict[str, Any]: if dry_run: return {"ok": True, "dry_run": True, "steps": plan} if not assume_yes: raise RuntimeError("real execution requires --yes") tat_client = build_tat_client(inventory) clb_client = build_clb_client(inventory) if any(step["clb_enabled"] for step in plan) else None results: list[dict[str, Any]] = [] for step in plan: started = time.time() record = dict(step) snapshot: dict[str, Any] | None = None try: if step["clb_enabled"]: snapshot = prepare_clb_snapshot(clb_client, step) set_clb_snapshot_weight(clb_client, inventory, snapshot, 0) if step["drain_seconds"] > 0: time.sleep(step["drain_seconds"]) record["clb_drained"] = True service_cfg = inventory["services"][step["service"]] script = remote_restart_script( service_cfg, image=step["image"] if step["action"] == "deploy" else "", health_timeout_seconds=int(inventory.get("service_health_timeout_seconds") or 150), ) remote = run_tat_shell( tat_client, instance_id=step["instance_id"], script=script, command_name=f"hyapp-{step['action']}-{step['service']}-{step['host']}", timeout_seconds=int(inventory.get("tat_timeout_seconds") or 900), poll_seconds=int(inventory.get("tat_poll_seconds") or 2), ) record["remote_status"] = remote["status"] record["remote_output"] = str(remote.get("output") or "")[-2000:] if remote["status"] != "SUCCESS": raise RuntimeError(f"TAT command failed: {remote['status']}") if snapshot is not None: # CLB 后端健康由 CLB 自己持续探测；服务本地健康通过后恢复原权重。 set_clb_snapshot_weight(clb_client, inventory, snapshot, None) record["clb_restored"] = True stabilize = int(inventory.get("stabilize_seconds") or 0) if stabilize > 0: time.sleep(stabilize) record["status"] = "success" record["duration_seconds"] = round(time.time() - started, 2) results.append(record) except Exception as exc: # noqa: BLE001 record["status"] = "failed" record["error"] = str(exc) record["duration_seconds"] = round(time.time() - started, 2) results.append(record) return {"ok": False, "steps": results, "error": str(exc)} return {"ok": True, "steps": results} def parse_args(argv: list[str]) -> argparse.Namespace: parser = argparse.ArgumentParser(description="Deploy hyapp services through Tencent TAT with optional CLB drain.") parser.add_argument("action", choices=["plan", "discover", "restart", "deploy"]) parser.add_argument("--inventory", default="deploy/tencent-tat/hyapp-services.inventory.prod.example.json") parser.add_argument("--env-file", default="", help="Optional .env file with Tencent Cloud credentials.") parser.add_argument("--services", required=True, help="Comma separated service names.") parser.add_argument("--hosts", default="", help="Optional comma separated host names for one-node maintenance.") parser.add_argument("--tag", default="", help="Release tag used by image_template during deploy.") parser.add_argument("--image", action="append", default=[], help="Override one image: service=image. Can be repeated.") parser.add_argument("--no-clb", action="store_true", help="Skip CLB drain. Only use for isolated emergency maintenance.") parser.add_argument("--dry-run", action="store_true", help="Print the execution plan without calling Tencent APIs.") parser.add_argument("--include-config", action="store_true", help="Include redacted service config YAML in discover output.") parser.add_argument("--max-config-bytes", type=int, default=12_000, help="Maximum bytes to read from each config file during discover.") parser.add_argument("--yes", action="store_true", help="Required for real restart/deploy execution.") return parser.parse_args(argv) def main(argv: list[str]) -> int: try: args = parse_args(argv) load_env_file(args.env_file) inventory = load_inventory(args.inventory) service_names = selected_service_names(inventory, args.services) overrides = image_overrides(args.image) action = "restart" if args.action == "plan" else args.action plan = build_plan( inventory, action=action, service_names=service_names, raw_hosts=args.hosts, tag=args.tag, overrides=overrides, no_clb=args.no_clb, ) if args.action == "plan": print(json.dumps({"ok": True, "steps": plan}, ensure_ascii=False, indent=2)) return 0 if args.action == "discover": result = execute_discovery( inventory, plan, dry_run=args.dry_run, include_config=args.include_config, max_config_bytes=args.max_config_bytes, ) print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if result.get("ok") else 1 result = execute_plan(inventory, plan, dry_run=args.dry_run, assume_yes=args.yes) print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if result.get("ok") else 1 except Exception as exc: # noqa: BLE001 print(json.dumps({"ok": False, "error": str(exc)}, ensure_ascii=False, indent=2), file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))