diff --git a/.env.example b/.env.example index fe79bfc..5ec598f 100644 --- a/.env.example +++ b/.env.example @@ -101,6 +101,11 @@ DEPLOY_REGION=me-saudi-arabia DEPLOY_INSTANCE_ID= TENCENT_SECRET_ID= TENCENT_SECRET_KEY= +GATEWAY_CLB_LOAD_BALANCER_ID= +GATEWAY_CLB_LISTENER_IDS= +GATEWAY_CLB_POLL_SECONDS=2 +GATEWAY_CLB_TASK_TIMEOUT_SECONDS=120 +GATEWAY_CLB_DRAIN_SECONDS=0 DEPLOY_SSH_HOST= DEPLOY_SSH_USER=root DEPLOY_SSH_PORT=22 diff --git a/README.md b/README.md index 586c365..66cf8a7 100644 --- a/README.md +++ b/README.md @@ -63,6 +63,11 @@ DEPLOY_REGION=me-saudi-arabia DEPLOY_INSTANCE_ID= TENCENT_SECRET_ID=*** TENCENT_SECRET_KEY=*** +GATEWAY_CLB_LOAD_BALANCER_ID=lb-xxxxxxxx +GATEWAY_CLB_LISTENER_IDS=lbl-http,lbl-https +GATEWAY_CLB_POLL_SECONDS=2 +GATEWAY_CLB_TASK_TIMEOUT_SECONDS=120 +GATEWAY_CLB_DRAIN_SECONDS=0 SSH_USER=root SSH_PORT=22 SSH_PRIVATE_KEY_PATH=/opt/hy-app-monitor/run/ssh/id_ed25519 @@ -260,6 +265,14 @@ python3 ops/sync_runtime_compose.py --sync-remote 6. 通过 `SSH` 先在目标机 `docker pull` 新镜像 7. 最后只对你选中的服务执行 `docker compose up -d --force-recreate --no-deps ` +如果选中的是 `gateway`，在真正重建单节点之前还会额外执行： + +1. 读取 `GATEWAY_CLB_LOAD_BALANCER_ID` + `GATEWAY_CLB_LISTENER_IDS` 下当前节点的全部后端绑定 +2. 把这些绑定的权重统一改成 `0`，避免新请求继续打到待重启节点 +3. 可选按 `GATEWAY_CLB_DRAIN_SECONDS` 额外等待连接自然排空 +4. 等实例本身 HTTP / Nacos 恢复后，再轮询 CLB 健康检查恢复 +5. 最后把每条绑定恢复到摘流前的原权重 + 如果选中 `admin` 前端目标，则会改走： 1. 在 deploy 机拉取 / 更新 `chatapp3-admin` @@ -273,6 +286,7 @@ python3 ops/sync_runtime_compose.py --sync-remote 说明： - `other` 现在也按 Harbor 业务镜像更新，不再继续依赖本地 `./other/service.jar` 挂载 +- `gateway` 发布现在强依赖 CLB 摘流配置；缺少 `GATEWAY_CLB_LOAD_BALANCER_ID` 或 `GATEWAY_CLB_LISTENER_IDS` 时，页面会给 warning，实际执行也会直接失败而不是冒险重启 - Java Maven 默认走增量 `package`，并支持通过 `JAVA_MAVEN_THREADS` 开启 reactor 并行 - 多个 Java 服务会按 `JAVA_IMAGE_BUILD_MAX_WORKERS` 受限并行执行 `docker build + push` - `admin` 前端不会走 Harbor 镜像链路，而是直接发布静态文件到 `site` diff --git a/core/config.py b/core/config.py index c3ec0ad..962ffcf 100644 --- a/core/config.py +++ b/core/config.py @@ -5,7 +5,7 @@ import os from pathlib import Path from typing import Any -from .env import env_float, env_int, env_int_list, env_str, load_env_file +from .env import env_float, env_int, env_int_list, env_str, env_str_list, load_env_file # 项目根目录。 @@ -114,6 +114,16 @@ DEPLOY_REGION = env_str("DEPLOY_REGION", "me-saudi-arabia") TENCENT_SECRET_ID = env_str("TENCENT_SECRET_ID", "") # 腾讯云 SecretKey。 TENCENT_SECRET_KEY = env_str("TENCENT_SECRET_KEY", "") +# gateway 发布做 CLB 摘流时使用的负载均衡实例 ID。 +GATEWAY_CLB_LOAD_BALANCER_ID = env_str("GATEWAY_CLB_LOAD_BALANCER_ID", "") +# gateway 可能同时挂在 HTTP / HTTPS 等多个监听器下，这里统一允许逗号分隔配置。 +GATEWAY_CLB_LISTENER_IDS = env_str_list("GATEWAY_CLB_LISTENER_IDS", []) +# CLB 异步任务轮询间隔。 +GATEWAY_CLB_POLL_SECONDS = env_float("GATEWAY_CLB_POLL_SECONDS", 2.0) +# CLB 摘流 / 恢复异步任务最长等待秒数。 +GATEWAY_CLB_TASK_TIMEOUT_SECONDS = env_int("GATEWAY_CLB_TASK_TIMEOUT_SECONDS", 120) +# 摘流后额外等待多少秒再重启；默认 0 表示只停新流量，不额外等待连接自然排空。 +GATEWAY_CLB_DRAIN_SECONDS = env_float("GATEWAY_CLB_DRAIN_SECONDS", 0.0) def pick_existing_path(candidates: list[Path]) -> Path: diff --git a/core/env.py b/core/env.py index b2a4f50..d3540ef 100644 --- a/core/env.py +++ b/core/env.py @@ -82,3 +82,18 @@ def env_int_list(name: str, default: list[int]) -> tuple[int, ...]: # 转成整数元组返回。 return tuple(int(item) for item in values) + + +def env_str_list(name: str, default: list[str]) -> tuple[str, ...]: + # 先读取原始文本。 + raw_value = env_str(name, "") + + # 没配置时直接返回默认值。 + if not raw_value: + return tuple(str(item).strip() for item in default if str(item).strip()) + + # 逐项切分并过滤空项。 + values = [item.strip() for item in raw_value.split(",") if item.strip()] + + # 返回去重后的字符串元组，避免调用方重复处理。 + return tuple(dict.fromkeys(values)) diff --git a/monitors/yumi/service_release.py b/monitors/yumi/service_release.py index 64eb113..837a750 100644 --- a/monitors/yumi/service_release.py +++ b/monitors/yumi/service_release.py @@ -42,6 +42,7 @@ from core.config import ( ADMIN_FRONTEND_SSH_USE_SUDO, ADMIN_FRONTEND_SSH_USER, BUILD_DOCKER_PLATFORM, + GATEWAY_CLB_DRAIN_SECONDS, GOLANG_REPO_ROOT, HARBOR_PASSWORD, HARBOR_PROJECT, @@ -69,6 +70,13 @@ from .service_ops import ( wait_java_registration_ready, ) from .ssh_remote import copy_local_path_to_host, read_remote_text, run_host_script_checked, target_display, write_remote_text +from .tencent_clb import ( + drain_gateway_release_target, + gateway_clb_configuration_error, + prepare_gateway_release_clb_snapshot, + restore_gateway_release_target, + wait_gateway_release_target_healthy, +) # Java 仓里当前支持 Harbor 更新的服务集合。 @@ -1467,6 +1475,10 @@ def build_update_targets_payload() -> dict[str, Any]: ], } # 某些主机模板暂时不可读时，仍然保留服务条目，只把原因透出去。 + if service_name == "gateway": + gateway_warning = gateway_clb_configuration_error() + if gateway_warning and gateway_warning not in compose_warnings: + compose_warnings.append(gateway_warning) if compose_warnings: item["warnings"] = compose_warnings items.append(item) @@ -1611,8 +1623,39 @@ def deploy_runtime_services_payload( "image": service_images[service_name], "status": "running", } + gateway_clb_snapshot: dict[str, Any] | None = None + gateway_clb_health_ready = False + gateway_clb_restored = False try: emit_progress(progress, step_stage, f"开始滚动更新：{service['host']} / {service_name}") + + if service_name == "gateway": + emit_progress(progress, step_stage, "读取 gateway 当前 CLB 绑定") + gateway_clb_snapshot = prepare_gateway_release_clb_snapshot(service) + step["clb"] = { + "loadBalancerId": gateway_clb_snapshot["loadBalancerId"], + "listenerIds": list(gateway_clb_snapshot["listenerIds"]), + "bindingCount": len(gateway_clb_snapshot["bindings"]), + "bindings": [ + { + "listenerId": binding["listenerId"], + "locationId": binding["locationId"], + "domain": binding["domain"], + "url": binding["url"], + "port": binding["port"], + "originalWeight": binding["weight"], + } + for binding in gateway_clb_snapshot["bindings"] + ], + } + emit_progress(progress, step_stage, f"CLB 摘流前权重：{gateway_clb_snapshot['bindingSummary']}") + drain_result = drain_gateway_release_target(gateway_clb_snapshot) + step["clb"]["drainedSummary"] = drain_result["summary"] + emit_progress(progress, step_stage, f"CLB 摘流完成：{drain_result['summary']}") + if GATEWAY_CLB_DRAIN_SECONDS > 0: + emit_progress(progress, step_stage, f"CLB 排空等待 {GATEWAY_CLB_DRAIN_SECONDS}s") + time.sleep(GATEWAY_CLB_DRAIN_SECONDS) + restart_result = restart_service_instance(service) step["remoteStatus"] = restart_result.get("status") step["remoteOutput"] = str(restart_result.get("output") or "").strip()[:500] @@ -1646,6 +1689,22 @@ def deploy_runtime_services_payload( f"Nacos 注册恢复：{step['nacos'].get('source') or 'ok'}", ) + if gateway_clb_snapshot is not None: + clb_health = wait_gateway_release_target_healthy( + gateway_clb_snapshot, + timeout_seconds=ROLLING_RESTART_TIMEOUT_SECONDS, + ) + gateway_clb_health_ready = True + step["clb"]["health"] = { + "ok": True, + "targets": clb_health, + } + emit_progress(progress, step_stage, "CLB 健康检查恢复，开始恢复转发权重") + restore_result = restore_gateway_release_target(gateway_clb_snapshot) + gateway_clb_restored = True + step["clb"]["restoredSummary"] = restore_result["summary"] + emit_progress(progress, step_stage, f"CLB 权重恢复完成：{restore_result['summary']}") + if ROLLING_RESTART_STABILIZE_SECONDS > 0: emit_progress( progress, @@ -1663,6 +1722,14 @@ def deploy_runtime_services_payload( f"实例更新成功，耗时 {step['durationSeconds']}s", ) except Exception as exc: # noqa: BLE001 + if gateway_clb_snapshot is not None and gateway_clb_health_ready and not gateway_clb_restored: + try: + restore_result = restore_gateway_release_target(gateway_clb_snapshot) + gateway_clb_restored = True + step.setdefault("clb", {})["restoreRecovered"] = restore_result["summary"] + emit_progress(progress, step_stage, f"失败收尾时已恢复 CLB 权重：{restore_result['summary']}") + except Exception as restore_exc: # noqa: BLE001 + step.setdefault("clb", {})["restoreError"] = str(restore_exc) step["status"] = "failed" step["error"] = str(exc) step["durationSeconds"] = round(time.time() - started_at, 2) diff --git a/monitors/yumi/tencent_clb.py b/monitors/yumi/tencent_clb.py new file mode 100644 index 0000000..1276b8c --- /dev/null +++ b/monitors/yumi/tencent_clb.py @@ -0,0 +1,439 @@ +from __future__ import annotations + +import json +import time +from typing import Any + +from core.config import ( + DEPLOY_REGION, + GATEWAY_CLB_LISTENER_IDS, + GATEWAY_CLB_LOAD_BALANCER_ID, + GATEWAY_CLB_POLL_SECONDS, + GATEWAY_CLB_TASK_TIMEOUT_SECONDS, + TENCENT_SECRET_ID, + TENCENT_SECRET_KEY, + utc_now, +) + + +def gateway_clb_configuration_error() -> str: + # gateway 发布一旦要走摘流，就必须拿到完整的腾讯云和 CLB 配置。 + missing: list[str] = [] + if not TENCENT_SECRET_ID: + missing.append("TENCENT_SECRET_ID") + if not TENCENT_SECRET_KEY: + missing.append("TENCENT_SECRET_KEY") + if not DEPLOY_REGION: + missing.append("DEPLOY_REGION") + if not GATEWAY_CLB_LOAD_BALANCER_ID: + missing.append("GATEWAY_CLB_LOAD_BALANCER_ID") + if not GATEWAY_CLB_LISTENER_IDS: + missing.append("GATEWAY_CLB_LISTENER_IDS") + + if missing: + return f"gateway CLB protection requires .env: {', '.join(missing)}" + return "" + + +def gateway_clb_settings() -> dict[str, Any]: + # 配置检查单独收口，发布页预警和真正执行都复用同一套口径。 + error = gateway_clb_configuration_error() + if error: + raise RuntimeError(error) + + return { + "loadBalancerId": GATEWAY_CLB_LOAD_BALANCER_ID, + "listenerIds": list(GATEWAY_CLB_LISTENER_IDS), + } + + +def build_clb_client() -> tuple[Any | None, str]: + # 先检查最基本的环境变量，缺任何一项都不继续请求腾讯云。 + error = gateway_clb_configuration_error() + if error: + return None, error + + try: + from tencentcloud.common import credential + from tencentcloud.common.profile.client_profile import ClientProfile + from tencentcloud.common.profile.http_profile import HttpProfile + from tencentcloud.clb.v20180317 import clb_client + except Exception as exc: # noqa: BLE001 + return None, f"tencentcloud sdk unavailable: {exc}" + + cred = credential.Credential(TENCENT_SECRET_ID, TENCENT_SECRET_KEY) + client = clb_client.ClbClient( + cred, + DEPLOY_REGION, + ClientProfile(httpProfile=HttpProfile(endpoint="clb.tencentcloudapi.com")), + ) + return client, "" + + +def invoke_clb(action: Any, label: str, retries: int = 3, delay_seconds: int = 2) -> Any: + # SDK 异常这里集中做简单重试，避免上层每一步都手写同样的退避逻辑。 + from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException + + last_error: Exception | None = None + for attempt in range(1, retries + 1): + try: + return action() + except TencentCloudSDKException as exc: + last_error = exc + if attempt == retries: + break + time.sleep(delay_seconds * attempt) + + if last_error is not None: + raise last_error + raise RuntimeError(f"{label} failed without error") + + +def call_clb(method_name: str, payload: dict[str, Any]) -> dict[str, Any]: + # CLB 调用统一走 json 请求体，后面加字段时不用反复和 SDK 模型属性打交道。 + client, error = build_clb_client() + if client is None: + raise RuntimeError(error) + + 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_clb(lambda: getattr(client, method_name)(request), method_name) + return dict(json.loads(response.to_json_string()).get("Response") or {}) + + +def wait_clb_task(task_id: str, *, timeout_seconds: int = GATEWAY_CLB_TASK_TIMEOUT_SECONDS) -> dict[str, Any]: + # 所有异步修改都轮询到终态，避免“接口返回成功但后端任务失败”的假成功。 + if not task_id: + raise RuntimeError("missing CLB task id") + + deadline = time.time() + max(float(timeout_seconds), 1.0) + last_payload: dict[str, Any] = {} + while time.time() < deadline: + last_payload = call_clb("DescribeTaskStatus", {"TaskId": task_id}) + status = int(last_payload.get("Status") if last_payload.get("Status") is not None else 2) + if status == 0: + return last_payload + if status == 1: + message = str(last_payload.get("Message") or "CLB task failed").strip() + raise RuntimeError(f"CLB task failed: {message}") + time.sleep(max(float(GATEWAY_CLB_POLL_SECONDS), 0.1)) + + raise RuntimeError(f"CLB task timed out: {task_id}") + + +def target_matches_service(target: dict[str, Any], service: dict[str, Any]) -> bool: + # 匹配优先认实例 ID，其次再用私网 IP 和端口兜底，避免误把另一台 gateway 当成当前节点。 + target_instance_id = str(target.get("InstanceId") or target.get("TargetId") or "").strip() + service_instance_id = str(service.get("instanceId") or "").strip() + if service_instance_id and target_instance_id and target_instance_id != service_instance_id: + return False + + target_port = int(target.get("Port") or 0) + service_port = int(service.get("port") or 0) + if service_port and target_port and target_port != service_port: + return False + + service_ip = str(service.get("ip") or "").strip() + target_ips = [str(item or "").strip() for item in (target.get("PrivateIpAddresses") or [target.get("IP")]) if str(item or "").strip()] + if service_ip and target_ips and service_ip not in target_ips: + return False + + return bool(target_instance_id or target_ips) + + +def binding_key(binding: dict[str, Any]) -> str: + # 同一实例可能同时挂在多个监听器/规则下，这里把绑定维度完整拼成稳定 key。 + return "|".join( + [ + str(binding.get("listenerId") or "").strip(), + str(binding.get("locationId") or "").strip(), + str(binding.get("instanceId") or "").strip(), + str(int(binding.get("port") or 0)), + ] + ) + + +def binding_route_label(binding: dict[str, Any]) -> str: + # 日志里优先显示规则 ID，没有规则 ID 时退回到监听器 ID。 + parts = [str(binding.get("listenerId") or "").strip()] + location_id = str(binding.get("locationId") or "").strip() + if location_id: + parts.append(location_id) + domain = str(binding.get("domain") or "").strip() + url = str(binding.get("url") or "").strip() + if domain or url: + parts.append(f"{domain or '*'}{url or '/'}") + return " / ".join(part for part in parts if part) + + +def build_binding(listener: dict[str, Any], service: dict[str, Any], target: dict[str, Any], *, rule: dict[str, Any] | None) -> dict[str, Any]: + # 摘流和恢复都只需要最小必要字段，这里统一整理成稳定结构。 + private_ips = [str(item or "").strip() for item in (target.get("PrivateIpAddresses") or []) if str(item or "").strip()] + binding = { + "listenerId": str(listener.get("ListenerId") or "").strip(), + "locationId": str((rule or {}).get("LocationId") or "").strip(), + "domain": str((rule or {}).get("Domain") or "").strip(), + "url": str((rule or {}).get("Url") or "").strip(), + "instanceId": str(target.get("InstanceId") or target.get("TargetId") or service.get("instanceId") or "").strip(), + "port": int(target.get("Port") or service.get("port") or 0), + "weight": int(target.get("Weight") or 0), + "privateIp": private_ips[0] if private_ips else str(service.get("ip") or "").strip(), + "routeLabel": "", + } + binding["routeLabel"] = binding_route_label(binding) + return binding + + +def collect_service_bindings(listeners: list[dict[str, Any]], service: dict[str, Any]) -> list[dict[str, Any]]: + # 自动扫描当前实例在配置监听器里的全部绑定，避免只摘一条规则导致仍有流量漏进来。 + bindings_by_key: dict[str, dict[str, Any]] = {} + + for listener in listeners: + for target in listener.get("Targets") or []: + if not target_matches_service(target, service): + continue + binding = build_binding(listener, service, target, rule=None) + bindings_by_key[binding_key(binding)] = binding + + for rule in listener.get("Rules") or []: + for target in rule.get("Targets") or []: + if not target_matches_service(target, service): + continue + binding = build_binding(listener, service, target, rule=rule) + bindings_by_key[binding_key(binding)] = binding + + return list(bindings_by_key.values()) + + +def describe_gateway_targets() -> dict[str, Any]: + # 直接按配置监听器拉绑定列表，摘流只关心这几个 listener 的后端状态。 + settings = gateway_clb_settings() + return call_clb( + "DescribeTargets", + { + "LoadBalancerId": settings["loadBalancerId"], + "ListenerIds": settings["listenerIds"], + }, + ) + + +def current_gateway_bindings(snapshot: dict[str, Any]) -> list[dict[str, Any]]: + # 每次修改后都重新读一遍 CLB 当前值，确认线上状态已经真正落地。 + payload = describe_gateway_targets() + return collect_service_bindings(list(payload.get("Listeners") or []), snapshot["service"]) + + +def render_weight_summary(bindings: list[dict[str, Any]]) -> str: + # 日志只保留规则和权重摘要，避免一条进度消息塞太多冗余字段。 + parts = [f"{binding['routeLabel']} -> {int(binding.get('weight') or 0)}" for binding in bindings] + return ", ".join(parts) + + +def prepare_gateway_release_clb_snapshot(service: dict[str, Any]) -> dict[str, Any]: + # 进入单节点重启前，先把这个节点当前在 CLB 里的全部绑定关系记下来。 + if str(service.get("service") or "").strip() != "gateway": + raise RuntimeError("gateway CLB protection only supports gateway service") + + settings = gateway_clb_settings() + listeners = list(describe_gateway_targets().get("Listeners") or []) + bindings = collect_service_bindings(listeners, service) + if not bindings: + raise RuntimeError( + f"gateway target not found in configured CLB listeners: {service.get('host')} {service.get('instanceId') or service.get('ip')}" + ) + + return { + "service": { + "host": str(service.get("host") or "").strip(), + "service": str(service.get("service") or "").strip(), + "ip": str(service.get("ip") or "").strip(), + "instanceId": str(service.get("instanceId") or "").strip(), + "port": int(service.get("port") or 0), + }, + "loadBalancerId": settings["loadBalancerId"], + "listenerIds": list(settings["listenerIds"]), + "bindings": bindings, + "bindingSummary": render_weight_summary(bindings), + "updatedAt": utc_now(), + } + + +def modify_gateway_binding_weight(binding: dict[str, Any], weight: int) -> dict[str, Any]: + # 单条绑定单独改权重，出错时能直接定位到具体监听器/规则。 + payload: dict[str, Any] = { + "LoadBalancerId": GATEWAY_CLB_LOAD_BALANCER_ID, + "ListenerId": binding["listenerId"], + "Targets": [ + { + "InstanceId": binding["instanceId"], + "Port": int(binding["port"]), + } + ], + "Weight": int(weight), + } + if str(binding.get("locationId") or "").strip(): + payload["LocationId"] = binding["locationId"] + elif str(binding.get("domain") or "").strip() or str(binding.get("url") or "").strip(): + payload["Domain"] = str(binding.get("domain") or "").strip() + payload["Url"] = str(binding.get("url") or "").strip() + + response = call_clb("ModifyTargetWeight", payload) + wait_clb_task(str(response.get("RequestId") or "").strip()) + return { + "routeLabel": binding["routeLabel"], + "weight": int(weight), + "updatedAt": utc_now(), + } + + +def wait_gateway_binding_weights(snapshot: dict[str, Any], expected_weights: dict[str, int], *, timeout_seconds: int) -> list[dict[str, Any]]: + # 权重调整是异步任务，必须读回线上值确认全部命中目标权重后才能继续重启。 + deadline = time.time() + max(float(timeout_seconds), 1.0) + last_bindings: list[dict[str, Any]] = [] + while time.time() < deadline: + last_bindings = current_gateway_bindings(snapshot) + current_by_key = {binding_key(item): item for item in last_bindings} + if current_by_key and all( + key in current_by_key and int(current_by_key[key].get("weight") or 0) == int(expected_weights[key]) + for key in expected_weights + ): + return last_bindings + time.sleep(max(float(GATEWAY_CLB_POLL_SECONDS), 0.1)) + + if not last_bindings: + raise RuntimeError("gateway CLB bindings disappeared while waiting for target weight") + + current_by_key = {binding_key(item): item for item in last_bindings} + raise RuntimeError( + "gateway CLB weight not ready: " + + ", ".join( + f"{current_by_key.get(key, {}).get('routeLabel', key)}={int(current_by_key.get(key, {}).get('weight') or -1)}" + for key in expected_weights + ) + ) + + +def drain_gateway_release_target(snapshot: dict[str, Any]) -> dict[str, Any]: + # 摘流直接把当前节点在所有匹配规则下的权重调成 0，避免新请求继续打到待重启节点。 + expected_weights: dict[str, int] = {} + operations: list[dict[str, Any]] = [] + for binding in snapshot["bindings"]: + operations.append(modify_gateway_binding_weight(binding, 0)) + expected_weights[binding_key(binding)] = 0 + + verified = wait_gateway_binding_weights( + snapshot, + expected_weights, + timeout_seconds=GATEWAY_CLB_TASK_TIMEOUT_SECONDS, + ) + return { + "ok": True, + "updatedAt": utc_now(), + "operations": operations, + "bindings": verified, + "summary": render_weight_summary(verified), + } + + +def restore_gateway_release_target(snapshot: dict[str, Any]) -> dict[str, Any]: + # 服务恢复后把每条绑定恢复成摘流前的原权重，避免线上流量分配被改坏。 + expected_weights: dict[str, int] = {} + operations: list[dict[str, Any]] = [] + for binding in snapshot["bindings"]: + original_weight = int(binding.get("weight") or 0) + operations.append(modify_gateway_binding_weight(binding, original_weight)) + expected_weights[binding_key(binding)] = original_weight + + verified = wait_gateway_binding_weights( + snapshot, + expected_weights, + timeout_seconds=GATEWAY_CLB_TASK_TIMEOUT_SECONDS, + ) + return { + "ok": True, + "updatedAt": utc_now(), + "operations": operations, + "bindings": verified, + "summary": render_weight_summary(verified), + } + + +def health_key(listener_id: str, location_id: str, target_id: str, port: int) -> str: + # 健康状态和绑定状态用同一套 key，便于直接按绑定反查对应健康条目。 + return "|".join([listener_id.strip(), location_id.strip(), target_id.strip(), str(int(port or 0))]) + + +def collect_gateway_health_states(load_balancers: list[dict[str, Any]], snapshot: dict[str, Any]) -> list[dict[str, Any]]: + # CLB 健康查询返回的是 loadBalancer -> listener -> rule -> target 结构，这里拍平后只留当前实例。 + matched: dict[str, dict[str, Any]] = {} + for load_balancer in load_balancers: + for listener in load_balancer.get("Listeners") or []: + listener_id = str(listener.get("ListenerId") or "").strip() + for rule in listener.get("Rules") or []: + location_id = str(rule.get("LocationId") or "").strip() + for target in rule.get("Targets") or []: + if not target_matches_service(target, snapshot["service"]): + continue + state = { + "listenerId": listener_id, + "locationId": location_id, + "targetId": str(target.get("TargetId") or "").strip(), + "port": int(target.get("Port") or 0), + "healthStatus": bool(target.get("HealthStatus")), + "detail": str(target.get("HealthStatusDetail") or target.get("HealthStatusDetial") or "").strip(), + } + matched[health_key(state["listenerId"], state["locationId"], state["targetId"], state["port"])] = state + return list(matched.values()) + + +def describe_gateway_target_health(snapshot: dict[str, Any]) -> list[dict[str, Any]]: + # 健康查询只拉当前负载均衡和监听器，避免把无关 listener 的状态混进来。 + payload = call_clb( + "DescribeTargetHealth", + { + "LoadBalancerIds": [snapshot["loadBalancerId"]], + "ListenerIds": list(snapshot["listenerIds"]), + }, + ) + return collect_gateway_health_states(list(payload.get("LoadBalancers") or []), snapshot) + + +def wait_gateway_release_target_healthy(snapshot: dict[str, Any], *, timeout_seconds: int) -> list[dict[str, Any]]: + # 只有等 CLB 自己也判定节点健康，才允许恢复权重并继续下一台，避免下一轮摘流时只剩空节点。 + expected_keys = { + health_key( + str(binding.get("listenerId") or "").strip(), + str(binding.get("locationId") or "").strip(), + str(binding.get("instanceId") or "").strip(), + int(binding.get("port") or 0), + ) + for binding in snapshot["bindings"] + } + deadline = time.time() + max(float(timeout_seconds), 1.0) + last_states: list[dict[str, Any]] = [] + while time.time() < deadline: + last_states = describe_gateway_target_health(snapshot) + current_by_key = { + health_key( + str(item.get("listenerId") or "").strip(), + str(item.get("locationId") or "").strip(), + str(item.get("targetId") or "").strip(), + int(item.get("port") or 0), + ): item + for item in last_states + } + if current_by_key and all( + key in current_by_key and bool(current_by_key[key].get("healthStatus")) + for key in expected_keys + ): + return last_states + time.sleep(max(float(GATEWAY_CLB_POLL_SECONDS), 0.1)) + + detail = ", ".join( + f"{item.get('listenerId')}/{item.get('locationId') or '-'}={item.get('detail') or item.get('healthStatus')}" + for item in last_states + ) + raise RuntimeError(f"gateway CLB health not ready: {detail or 'no matching target health'}") diff --git a/tests/test_gateway_release_clb.py b/tests/test_gateway_release_clb.py new file mode 100644 index 0000000..4f0c542 --- /dev/null +++ b/tests/test_gateway_release_clb.py @@ -0,0 +1,407 @@ +from __future__ import annotations + +import unittest +from unittest.mock import patch + +from monitors.yumi.service_release import deploy_runtime_services_payload +from monitors.yumi.tencent_clb import collect_service_bindings, prepare_gateway_release_clb_snapshot + + +class GatewayClbDiscoveryTests(unittest.TestCase): + @patch("monitors.yumi.tencent_clb.GATEWAY_CLB_LOAD_BALANCER_ID", "lb-test") + @patch("monitors.yumi.tencent_clb.GATEWAY_CLB_LISTENER_IDS", ("lbl-http", "lbl-https")) + @patch("monitors.yumi.tencent_clb.describe_gateway_targets") + def test_prepare_gateway_release_clb_snapshot_collects_all_matching_rules( + self, + describe_gateway_targets_mock, + ) -> None: + describe_gateway_targets_mock.return_value = { + "Listeners": [ + { + "ListenerId": "lbl-http", + "Rules": [ + { + "LocationId": "loc-a", + "Domain": "api.example.com", + "Url": "/", + "Targets": [ + { + "InstanceId": "ins-gw-1", + "Port": 8080, + "Weight": 20, + "PrivateIpAddresses": ["10.0.0.1"], + } + ], + }, + { + "LocationId": "loc-b", + "Domain": "admin.example.com", + "Url": "/", + "Targets": [ + { + "InstanceId": "ins-gw-1", + "Port": 8080, + "Weight": 10, + "PrivateIpAddresses": ["10.0.0.1"], + } + ], + }, + ], + }, + { + "ListenerId": "lbl-https", + "Rules": [ + { + "LocationId": "loc-c", + "Domain": "api.example.com", + "Url": "/", + "Targets": [ + { + "InstanceId": "ins-gw-2", + "Port": 8080, + "Weight": 15, + "PrivateIpAddresses": ["10.0.0.2"], + } + ], + } + ], + }, + ] + } + + snapshot = prepare_gateway_release_clb_snapshot( + { + "host": "gateway-1", + "service": "gateway", + "ip": "10.0.0.1", + "instanceId": "ins-gw-1", + "port": 8080, + } + ) + + self.assertEqual(snapshot["loadBalancerId"], "lb-test") + self.assertEqual(snapshot["listenerIds"], ["lbl-http", "lbl-https"]) + self.assertEqual( + {(item["listenerId"], item["locationId"], item["weight"]) for item in snapshot["bindings"]}, + {("lbl-http", "loc-a", 20), ("lbl-http", "loc-b", 10)}, + ) + + def test_collect_service_bindings_ignores_other_targets(self) -> None: + listeners = [ + { + "ListenerId": "lbl-http", + "Rules": [ + { + "LocationId": "loc-a", + "Targets": [ + { + "InstanceId": "ins-gw-1", + "Port": 8080, + "Weight": 20, + "PrivateIpAddresses": ["10.0.0.1"], + }, + { + "InstanceId": "ins-gw-2", + "Port": 8080, + "Weight": 20, + "PrivateIpAddresses": ["10.0.0.2"], + }, + ], + } + ], + } + ] + + bindings = collect_service_bindings( + listeners, + { + "host": "gateway-1", + "service": "gateway", + "ip": "10.0.0.1", + "instanceId": "ins-gw-1", + "port": 8080, + }, + ) + + self.assertEqual(len(bindings), 1) + self.assertEqual(bindings[0]["instanceId"], "ins-gw-1") + + +class GatewayReleaseFlowTests(unittest.TestCase): + @patch("monitors.yumi.service_release.build_service_images") + @patch("monitors.yumi.service_release.service_records_by_name") + @patch("monitors.yumi.service_release.buildable_service_names", return_value=["gateway", "auth"]) + @patch("monitors.yumi.service_release.host_image_plan") + @patch("monitors.yumi.service_release.update_local_service_image") + @patch("monitors.yumi.service_release.sync_remote_host_compose_template") + @patch("monitors.yumi.service_release.prepull_host_images") + @patch("monitors.yumi.service_release.prepare_gateway_release_clb_snapshot") + @patch("monitors.yumi.service_release.drain_gateway_release_target") + @patch("monitors.yumi.service_release.restart_service_instance") + @patch("monitors.yumi.service_release.wait_http_ready") + @patch("monitors.yumi.service_release.wait_java_registration_ready") + @patch("monitors.yumi.service_release.wait_gateway_release_target_healthy") + @patch("monitors.yumi.service_release.restore_gateway_release_target") + def test_gateway_update_drains_before_restart_and_restores_after_health( + self, + restore_gateway_release_target_mock, + wait_gateway_release_target_healthy_mock, + wait_java_registration_ready_mock, + wait_http_ready_mock, + restart_service_instance_mock, + drain_gateway_release_target_mock, + prepare_gateway_release_clb_snapshot_mock, + prepull_host_images_mock, + sync_remote_host_compose_template_mock, + update_local_service_image_mock, + host_image_plan_mock, + buildable_service_names_mock, + service_records_by_name_mock, + build_service_images_mock, + ) -> None: + del buildable_service_names_mock + gateway_record = { + "host": "gateway-1", + "ip": "10.0.0.1", + "instanceId": "ins-gw-1", + "port": 8080, + "path": "/health", + "service": "gateway", + "kind": "java", + } + service_records_by_name_mock.return_value = {"gateway": [gateway_record]} + build_service_images_mock.return_value = { + "serviceImages": {"gateway": "registry/gateway:new"}, + "builds": [], + } + host_image_plan_mock.return_value = [ + { + "host": "gateway-1", + "ip": "10.0.0.1", + "services": ["gateway"], + "images": ["registry/gateway:new"], + } + ] + update_local_service_image_mock.return_value = {"host": "gateway-1", "service": "gateway"} + sync_remote_host_compose_template_mock.return_value = { + "path": "/tmp/docker-compose.yml", + "changed": False, + "source": "local", + "backupPath": "", + } + prepull_host_images_mock.return_value = {"status": "SUCCESS", "output": "ok"} + prepare_gateway_release_clb_snapshot_mock.return_value = { + "loadBalancerId": "lb-test", + "listenerIds": ["lbl-http"], + "bindings": [{"listenerId": "lbl-http", "locationId": "loc-a", "domain": "", "url": "", "port": 8080, "weight": 10}], + "bindingSummary": "lbl-http / loc-a -> 10", + } + drain_gateway_release_target_mock.return_value = {"summary": "lbl-http / loc-a -> 0"} + restart_service_instance_mock.return_value = {"status": "SUCCESS", "output": "restarted"} + wait_http_ready_mock.return_value = {"ok": True, "statusCode": 200, "latencyMs": 12, "detail": "ok"} + wait_java_registration_ready_mock.return_value = { + "checked": True, + "healthy": True, + "enabled": True, + "source": "instance/list", + } + wait_gateway_release_target_healthy_mock.return_value = [{"listenerId": "lbl-http", "locationId": "loc-a", "healthStatus": True}] + restore_gateway_release_target_mock.return_value = {"summary": "lbl-http / loc-a -> 10"} + + call_order: list[str] = [] + + prepare_gateway_release_clb_snapshot_mock.side_effect = lambda service: call_order.append("prepare") or { + "loadBalancerId": "lb-test", + "listenerIds": ["lbl-http"], + "bindings": [{"listenerId": "lbl-http", "locationId": "loc-a", "domain": "", "url": "", "port": 8080, "weight": 10}], + "bindingSummary": "lbl-http / loc-a -> 10", + } + drain_gateway_release_target_mock.side_effect = lambda snapshot: call_order.append("drain") or {"summary": "lbl-http / loc-a -> 0"} + restart_service_instance_mock.side_effect = lambda service: call_order.append("restart") or {"status": "SUCCESS", "output": "restarted"} + wait_http_ready_mock.side_effect = lambda service, timeout_seconds: call_order.append("http") or { + "ok": True, + "statusCode": 200, + "latencyMs": 12, + "detail": "ok", + } + wait_java_registration_ready_mock.side_effect = lambda service, timeout_seconds: call_order.append("nacos") or { + "checked": True, + "healthy": True, + "enabled": True, + "source": "instance/list", + } + wait_gateway_release_target_healthy_mock.side_effect = lambda snapshot, timeout_seconds: call_order.append("clb-health") or [ + {"listenerId": "lbl-http", "locationId": "loc-a", "healthStatus": True} + ] + restore_gateway_release_target_mock.side_effect = lambda snapshot: call_order.append("restore") or { + "summary": "lbl-http / loc-a -> 10" + } + + result = deploy_runtime_services_payload( + ["gateway"], + java_git_ref="main", + go_git_ref="main", + image_tag="20260423", + skip_maven_build=True, + ) + + self.assertTrue(result["ok"]) + self.assertEqual(call_order, ["prepare", "drain", "restart", "http", "nacos", "clb-health", "restore"]) + self.assertEqual(result["steps"][0]["clb"]["restoredSummary"], "lbl-http / loc-a -> 10") + + @patch("monitors.yumi.service_release.build_service_images") + @patch("monitors.yumi.service_release.service_records_by_name") + @patch("monitors.yumi.service_release.buildable_service_names", return_value=["gateway"]) + @patch("monitors.yumi.service_release.host_image_plan") + @patch("monitors.yumi.service_release.update_local_service_image") + @patch("monitors.yumi.service_release.sync_remote_host_compose_template") + @patch("monitors.yumi.service_release.prepull_host_images") + @patch("monitors.yumi.service_release.prepare_gateway_release_clb_snapshot") + @patch("monitors.yumi.service_release.drain_gateway_release_target") + @patch("monitors.yumi.service_release.restart_service_instance", side_effect=RuntimeError("restart failed")) + @patch("monitors.yumi.service_release.restore_gateway_release_target") + def test_gateway_update_keeps_target_drained_when_restart_fails( + self, + restore_gateway_release_target_mock, + restart_service_instance_mock, + drain_gateway_release_target_mock, + prepare_gateway_release_clb_snapshot_mock, + prepull_host_images_mock, + sync_remote_host_compose_template_mock, + update_local_service_image_mock, + host_image_plan_mock, + buildable_service_names_mock, + service_records_by_name_mock, + build_service_images_mock, + ) -> None: + del buildable_service_names_mock, restart_service_instance_mock + gateway_record = { + "host": "gateway-1", + "ip": "10.0.0.1", + "instanceId": "ins-gw-1", + "port": 8080, + "path": "/health", + "service": "gateway", + "kind": "java", + } + service_records_by_name_mock.return_value = {"gateway": [gateway_record]} + build_service_images_mock.return_value = { + "serviceImages": {"gateway": "registry/gateway:new"}, + "builds": [], + } + host_image_plan_mock.return_value = [ + { + "host": "gateway-1", + "ip": "10.0.0.1", + "services": ["gateway"], + "images": ["registry/gateway:new"], + } + ] + update_local_service_image_mock.return_value = {"host": "gateway-1", "service": "gateway"} + sync_remote_host_compose_template_mock.return_value = { + "path": "/tmp/docker-compose.yml", + "changed": False, + "source": "local", + "backupPath": "", + } + prepull_host_images_mock.return_value = {"status": "SUCCESS", "output": "ok"} + prepare_gateway_release_clb_snapshot_mock.return_value = { + "loadBalancerId": "lb-test", + "listenerIds": ["lbl-http"], + "bindings": [{"listenerId": "lbl-http", "locationId": "loc-a", "domain": "", "url": "", "port": 8080, "weight": 10}], + "bindingSummary": "lbl-http / loc-a -> 10", + } + drain_gateway_release_target_mock.return_value = {"summary": "lbl-http / loc-a -> 0"} + + result = deploy_runtime_services_payload( + ["gateway"], + java_git_ref="main", + go_git_ref="main", + image_tag="20260423", + skip_maven_build=True, + ) + + self.assertFalse(result["ok"]) + self.assertIn("restart failed", result["error"]) + restore_gateway_release_target_mock.assert_not_called() + self.assertEqual(result["steps"][0]["clb"]["drainedSummary"], "lbl-http / loc-a -> 0") + + @patch("monitors.yumi.service_release.build_service_images") + @patch("monitors.yumi.service_release.service_records_by_name") + @patch("monitors.yumi.service_release.buildable_service_names", return_value=["auth"]) + @patch("monitors.yumi.service_release.host_image_plan") + @patch("monitors.yumi.service_release.update_local_service_image") + @patch("monitors.yumi.service_release.sync_remote_host_compose_template") + @patch("monitors.yumi.service_release.prepull_host_images") + @patch("monitors.yumi.service_release.prepare_gateway_release_clb_snapshot") + @patch("monitors.yumi.service_release.restart_service_instance") + @patch("monitors.yumi.service_release.wait_http_ready") + @patch("monitors.yumi.service_release.wait_java_registration_ready") + def test_non_gateway_update_skips_clb_protection( + self, + wait_java_registration_ready_mock, + wait_http_ready_mock, + restart_service_instance_mock, + prepare_gateway_release_clb_snapshot_mock, + prepull_host_images_mock, + sync_remote_host_compose_template_mock, + update_local_service_image_mock, + host_image_plan_mock, + buildable_service_names_mock, + service_records_by_name_mock, + build_service_images_mock, + ) -> None: + del buildable_service_names_mock + auth_record = { + "host": "app-1", + "ip": "10.0.0.3", + "instanceId": "ins-app-1", + "port": 1000, + "path": "/actuator/health", + "service": "auth", + "kind": "java", + } + service_records_by_name_mock.return_value = {"auth": [auth_record]} + build_service_images_mock.return_value = { + "serviceImages": {"auth": "registry/auth:new"}, + "builds": [], + } + host_image_plan_mock.return_value = [ + { + "host": "app-1", + "ip": "10.0.0.3", + "services": ["auth"], + "images": ["registry/auth:new"], + } + ] + update_local_service_image_mock.return_value = {"host": "app-1", "service": "auth"} + sync_remote_host_compose_template_mock.return_value = { + "path": "/tmp/docker-compose.yml", + "changed": False, + "source": "local", + "backupPath": "", + } + prepull_host_images_mock.return_value = {"status": "SUCCESS", "output": "ok"} + restart_service_instance_mock.return_value = {"status": "SUCCESS", "output": "restarted"} + wait_http_ready_mock.return_value = {"ok": True, "statusCode": 200, "latencyMs": 11, "detail": "ok"} + wait_java_registration_ready_mock.return_value = { + "checked": True, + "healthy": True, + "enabled": True, + "source": "instance/list", + } + + result = deploy_runtime_services_payload( + ["auth"], + java_git_ref="main", + go_git_ref="main", + image_tag="20260423", + skip_maven_build=True, + ) + + self.assertTrue(result["ok"]) + prepare_gateway_release_clb_snapshot_mock.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_service_release.py b/tests/test_service_release.py index c45cb70..90d0722 100644 --- a/tests/test_service_release.py +++ b/tests/test_service_release.py @@ -8,6 +8,47 @@ from monitors.yumi.service_release import build_update_targets_payload class ServiceReleaseTargetsTests(unittest.TestCase): + @patch("monitors.yumi.service_release.gateway_clb_configuration_error", return_value="gateway CLB protection requires .env") + @patch("monitors.yumi.service_release.frontend_release_enabled", return_value=False) + @patch("monitors.yumi.service_release.service_repository_ref", side_effect=lambda service: f"repo/{service}") + @patch("monitors.yumi.service_release.buildable_service_names", return_value=["gateway"]) + @patch("monitors.yumi.service_release.service_records_by_name") + @patch("monitors.yumi.service_release.load_local_host_compose_model") + def test_build_update_targets_payload_adds_gateway_clb_warning( + self, + load_local_host_compose_model_mock, + service_records_by_name_mock, + buildable_service_names_mock, + service_repository_ref_mock, + frontend_release_enabled_mock, + gateway_clb_configuration_error_mock, + ) -> None: + del buildable_service_names_mock + del service_repository_ref_mock + del frontend_release_enabled_mock + del gateway_clb_configuration_error_mock + service_records_by_name_mock.return_value = { + "gateway": [ + { + "host": "gateway-1", + "ip": "10.0.0.1", + "instanceId": "ins-gateway-1", + "port": 8080, + "path": "/health", + } + ] + } + load_local_host_compose_model_mock.return_value = { + "services": { + "gateway": {"image": "registry/gateway:main"}, + } + } + + payload = build_update_targets_payload() + + self.assertTrue(payload["ok"]) + self.assertEqual(payload["items"][0]["warnings"], ["gateway CLB protection requires .env"]) + @patch("monitors.yumi.service_release.frontend_release_enabled", return_value=False) @patch("monitors.yumi.service_release.service_repository_ref", side_effect=lambda service: f"repo/{service}") @patch("monitors.yumi.service_release.buildable_service_names", return_value=["auth", "gateway"])