From f89ebe797e11c854f0bbcf50cb6558b9039a6575 Mon Sep 17 00:00:00 2001 From: zhx Date: Fri, 10 Jul 2026 12:56:27 +0800 Subject: [PATCH] fix: drain internal CLB targets before restarting service instances MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 部署/滚动重启非 gateway 实例时，先把该实例在内网 CLB（app-internal/pay-internal） 上的后端权重摘到 0 并等待排空，实例恢复健康后再还原权重。 此前只做 Nacos 摘流，但业务服务之间（如 live -> other）经内网 CLB 固定 VIP 互调，Nacos 摘流对这部分流量无效：容器被杀后 CLB 健康检查（5s x 3 次）摘除 死后端之前，调用方请求持续打到死后端，把调用方线程池拖垮，表现为 “部署 other 时 live 挂掉一个”（health 探测超时 60-70 秒）。 通过 INTERNAL_CLB_IDS 配置启用；未配置时保持历史行为。 Co-Authored-By: Claude Fable 5 --- .env.example | 5 + core/config.py | 7 + monitors/yumi/service_ops.py | 37 ++++ monitors/yumi/service_release.py | 57 ++++++ monitors/yumi/tencent_clb.py | 311 ++++++++++++++++++++++++++++++- tests/test_internal_clb_drain.py | 127 +++++++++++++ 6 files changed, 541 insertions(+), 3 deletions(-) create mode 100644 tests/test_internal_clb_drain.py diff --git a/.env.example b/.env.example index b411392..23532b5 100644 --- a/.env.example +++ b/.env.example @@ -129,6 +129,11 @@ CDN_PURGE_TASK_LIMIT=20 GATEWAY_CLB_POLL_SECONDS=2 GATEWAY_CLB_TASK_TIMEOUT_SECONDS=120 GATEWAY_CLB_DRAIN_SECONDS=0 +# 业务服务互调走的内网 CLB（如 app-internal / pay-internal）。 +# 配置后，发布/滚动重启任何非 gateway 实例前会先把它在这些 CLB 上的后端权重摘到 0， +# 恢复健康后再还原权重；不配置则跳过内网 CLB 摘流（历史行为）。 +INTERNAL_CLB_IDS= +INTERNAL_CLB_DRAIN_SECONDS=10 DEPLOY_SSH_HOST= DEPLOY_SSH_USER=root DEPLOY_SSH_PORT=22 diff --git a/core/config.py b/core/config.py index 60acb12..4e678d2 100644 --- a/core/config.py +++ b/core/config.py @@ -136,6 +136,13 @@ GATEWAY_CLB_POLL_SECONDS = env_float("GATEWAY_CLB_POLL_SECONDS", 2.0) 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) +# 业务服务之间经内网 CLB 互调（feign 固定 VIP，如 live -> 10.2.1.16:2400 -> other）。 +# 发布前必须把当前实例在这些 CLB 上的后端权重摘到 0，否则 CLB 健康检查（默认 5s x 3 次） +# 摘除死后端之前，调用方（如 live）的请求会持续打到正在重启的节点并拖垮其线程池。 +# 留空表示不启用内网 CLB 摘流。 +INTERNAL_CLB_IDS = env_str_list("INTERNAL_CLB_IDS", []) +# 内网 CLB 权重摘到 0 并确认后，额外等待存量连接排空的秒数。 +INTERNAL_CLB_DRAIN_SECONDS = env_float("INTERNAL_CLB_DRAIN_SECONDS", 10.0) def pick_existing_path(candidates: list[Path]) -> Path: diff --git a/monitors/yumi/service_ops.py b/monitors/yumi/service_ops.py index 958a0d5..af09308 100644 --- a/monitors/yumi/service_ops.py +++ b/monitors/yumi/service_ops.py @@ -32,6 +32,12 @@ from .nacos import ( set_instance_enabled, ) from .ssh_remote import run_host_script +from .tencent_clb import ( + drain_internal_clb_target, + prepare_internal_clb_snapshot, + restore_internal_clb_target, + wait_internal_clb_target_healthy, +) def restartable_service_records() -> list[dict[str, Any]]: @@ -336,6 +342,9 @@ def rolling_restart_payload(service_names: list[str]) -> dict[str, Any]: "status": "running", } nacos_drained = False + internal_clb_snapshot: dict[str, Any] | None = None + internal_clb_drained = False + internal_clb_restored = False try: # 单机第一次进入重启前，先把完整 compose 模板同步到远端。 @@ -350,6 +359,19 @@ def rolling_restart_payload(service_names: list[str]) -> dict[str, Any]: "source": str(prepared_hosts[service["host"]].get("source") or ""), } + # 业务服务之间经内网 CLB 互调，先把当前实例权重摘到 0， + # 避免调用方（如 live）的请求打到正在重启的节点。gateway 走公网 CLB 专属链路。 + if service_name != "gateway": + internal_clb_snapshot = prepare_internal_clb_snapshot(service) + if internal_clb_snapshot is not None: + drain_result = drain_internal_clb_target(internal_clb_snapshot) + internal_clb_drained = True + step["internalClb"] = { + "bindingSummary": internal_clb_snapshot["bindingSummary"], + "drainedSummary": drain_result["summary"], + "drainSeconds": drain_result["drainSeconds"], + } + if service["kind"] == "java": drain_result = drain_java_registration(service) if drain_result.get("checked"): @@ -375,6 +397,15 @@ def rolling_restart_payload(service_names: list[str]) -> dict[str, Any]: else wait_java_registration_ready(service, ROLLING_RESTART_TIMEOUT_SECONDS) ) + if internal_clb_drained and internal_clb_snapshot is not None: + wait_internal_clb_target_healthy( + internal_clb_snapshot, + timeout_seconds=ROLLING_RESTART_TIMEOUT_SECONDS, + ) + restore_result = restore_internal_clb_target(internal_clb_snapshot) + internal_clb_restored = True + step["internalClb"]["restoredSummary"] = restore_result["summary"] + if ROLLING_RESTART_STABILIZE_SECONDS > 0: time.sleep(ROLLING_RESTART_STABILIZE_SECONDS) @@ -382,6 +413,12 @@ def rolling_restart_payload(service_names: list[str]) -> dict[str, Any]: step["durationSeconds"] = round(time.time() - started_at, 2) steps.append(step) except Exception as exc: # noqa: BLE001 + if internal_clb_drained and not internal_clb_restored and internal_clb_snapshot is not None: + try: + restore_result = restore_internal_clb_target(internal_clb_snapshot) + step.setdefault("internalClb", {})["restoreRecovered"] = restore_result["summary"] + except Exception as restore_exc: # noqa: BLE001 + step.setdefault("internalClb", {})["restoreError"] = str(restore_exc) if nacos_drained: try: step["nacosRestoreRecovered"] = update_java_registration_enabled(service, True) diff --git a/monitors/yumi/service_release.py b/monitors/yumi/service_release.py index 59654b4..4c3a2dd 100644 --- a/monitors/yumi/service_release.py +++ b/monitors/yumi/service_release.py @@ -82,10 +82,14 @@ from .service_ops import ( 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, + drain_internal_clb_target, gateway_clb_configuration_error, prepare_gateway_release_clb_snapshot, + prepare_internal_clb_snapshot, restore_gateway_release_target, + restore_internal_clb_target, wait_gateway_release_target_healthy, + wait_internal_clb_target_healthy, ) @@ -1870,9 +1874,35 @@ def deploy_runtime_services_payload( gateway_clb_health_ready = False gateway_clb_restored = False nacos_drained = False + internal_clb_snapshot: dict[str, Any] | None = None + internal_clb_drained = False + internal_clb_restored = False try: emit_progress(progress, step_stage, f"开始滚动更新：{service['host']} / {service_name}") + # 业务服务之间经内网 CLB（feign 固定 VIP）互调，Nacos 摘流对这部分流量无效。 + # 重启前先把当前实例在内网 CLB 上的权重摘到 0，避免调用方（如 live）打到死后端拖垮线程池。 + if service_name != "gateway": + internal_clb_snapshot = prepare_internal_clb_snapshot(service) + if internal_clb_snapshot is not None: + emit_progress( + progress, + step_stage, + f"内网 CLB 摘流前权重：{internal_clb_snapshot['bindingSummary']}", + ) + internal_drain_result = drain_internal_clb_target(internal_clb_snapshot) + internal_clb_drained = True + step["internalClb"] = { + "bindingSummary": internal_clb_snapshot["bindingSummary"], + "drainedSummary": internal_drain_result["summary"], + "drainSeconds": internal_drain_result["drainSeconds"], + } + emit_progress( + progress, + step_stage, + f"内网 CLB 摘流完成：{internal_drain_result['summary']} drain={internal_drain_result['drainSeconds']}s", + ) + if service_name == "gateway": emit_progress(progress, step_stage, "读取 gateway 当前 CLB 绑定") gateway_clb_snapshot = prepare_gateway_release_clb_snapshot(service) @@ -1963,6 +1993,21 @@ def deploy_runtime_services_payload( step["clb"]["restoredSummary"] = restore_result["summary"] emit_progress(progress, step_stage, f"CLB 权重恢复完成：{restore_result['summary']}") + if internal_clb_drained and internal_clb_snapshot is not None: + wait_internal_clb_target_healthy( + internal_clb_snapshot, + timeout_seconds=ROLLING_RESTART_TIMEOUT_SECONDS, + ) + emit_progress(progress, step_stage, "内网 CLB 健康检查恢复，开始恢复转发权重") + internal_restore_result = restore_internal_clb_target(internal_clb_snapshot) + internal_clb_restored = True + step["internalClb"]["restoredSummary"] = internal_restore_result["summary"] + emit_progress( + progress, + step_stage, + f"内网 CLB 权重恢复完成：{internal_restore_result['summary']}", + ) + if ROLLING_RESTART_STABILIZE_SECONDS > 0: emit_progress( progress, @@ -1980,6 +2025,18 @@ def deploy_runtime_services_payload( f"实例更新成功，耗时 {step['durationSeconds']}s", ) except Exception as exc: # noqa: BLE001 + if internal_clb_drained and not internal_clb_restored and internal_clb_snapshot is not None: + try: + internal_restore_result = restore_internal_clb_target(internal_clb_snapshot) + internal_clb_restored = True + step.setdefault("internalClb", {})["restoreRecovered"] = internal_restore_result["summary"] + emit_progress( + progress, + step_stage, + f"失败收尾时已恢复内网 CLB 权重：{internal_restore_result['summary']}", + ) + except Exception as restore_exc: # noqa: BLE001 + step.setdefault("internalClb", {})["restoreError"] = str(restore_exc) 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) diff --git a/monitors/yumi/tencent_clb.py b/monitors/yumi/tencent_clb.py index 4e9c59b..fb5adaf 100644 --- a/monitors/yumi/tencent_clb.py +++ b/monitors/yumi/tencent_clb.py @@ -11,6 +11,8 @@ from core.config import ( GATEWAY_CLB_LOAD_BALANCER_ID, GATEWAY_CLB_POLL_SECONDS, GATEWAY_CLB_TASK_TIMEOUT_SECONDS, + INTERNAL_CLB_DRAIN_SECONDS, + INTERNAL_CLB_IDS, TENCENT_SECRET_ID, TENCENT_SECRET_KEY, utc_now, @@ -49,9 +51,24 @@ def gateway_clb_settings() -> dict[str, Any]: } -def build_clb_client() -> tuple[Any | None, str]: - # 先检查最基本的环境变量，缺任何一项都不继续请求腾讯云。 - error = gateway_clb_configuration_error() +def clb_credentials_error() -> str: + # 内网 CLB 摘流只依赖腾讯云凭据和区域，不要求 gateway 专属配置。 + 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 missing: + return f"CLB api requires .env: {', '.join(missing)}" + return "" + + +def build_clb_api_client() -> tuple[Any | None, str]: + # 只做凭据级检查的通用 CLB 客户端，gateway 和内网摘流共用。 + error = clb_credentials_error() if error: return None, error @@ -72,6 +89,15 @@ def build_clb_client() -> tuple[Any | None, str]: return client, "" +def build_clb_client() -> tuple[Any | None, str]: + # 先检查最基本的环境变量，缺任何一项都不继续请求腾讯云。 + error = gateway_clb_configuration_error() + if error: + return None, error + + return build_clb_api_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 @@ -91,6 +117,24 @@ def invoke_clb(action: Any, label: str, retries: int = 3, delay_seconds: int = 2 raise RuntimeError(f"{label} failed without error") +def call_clb_api(method_name: str, payload: dict[str, Any]) -> dict[str, Any]: + # 通用 CLB 调用，只要求腾讯云凭据；内网 CLB 摘流走这里。 + client, error = build_clb_api_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) + response_payload = json.loads(response.to_json_string()) + if isinstance(response_payload, dict) and isinstance(response_payload.get("Response"), dict): + return dict(response_payload.get("Response") or {}) + return dict(response_payload or {}) + + def call_clb(method_name: str, payload: dict[str, Any]) -> dict[str, Any]: # CLB 调用统一走 json 请求体，后面加字段时不用反复和 SDK 模型属性打交道。 client, error = build_clb_client() @@ -461,3 +505,264 @@ def wait_gateway_release_target_healthy(snapshot: dict[str, Any], *, timeout_sec for item in last_states ) raise RuntimeError(f"gateway CLB health not ready: {detail or 'no matching target health'}") + + +# --------------------------------------------------------------------------- +# 内网 CLB 摘流：业务服务（other/live/wallet 等）之间经内网 CLB 固定 VIP 互调， +# Nacos 摘流对这部分流量无效。重启实例前必须把它在内网 CLB 上的后端权重摘到 0， +# 否则 CLB 健康检查（5s x 3 次）摘除死后端之前，调用方的请求会持续打到正在 +# 重启的节点，把调用方（如 live）的工作线程池拖垮，表现为“部署 other 时 live 挂掉”。 +# --------------------------------------------------------------------------- + + +def internal_clb_enabled() -> bool: + # 未配置 INTERNAL_CLB_IDS 时整个内网摘流链路静默关闭，行为与历史版本一致。 + return bool(INTERNAL_CLB_IDS) + + +def wait_clb_api_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_api("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 collect_internal_bindings(listeners: list[dict[str, Any]], service: dict[str, Any]) -> list[dict[str, Any]]: + # 内网 CLB 以 L4 监听器为主（Targets 直接挂在 listener 下），同时兼容规则型监听器。 + 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 prepare_internal_clb_snapshot(service: dict[str, Any]) -> dict[str, Any] | None: + # 扫描配置的内网 CLB，记录当前实例（ip+port）挂载的全部后端及原权重。 + # 服务不在任何内网 CLB 后端时返回 None，调用方按无需摘流处理。 + if not internal_clb_enabled(): + return None + + groups: list[dict[str, Any]] = [] + for load_balancer_id in INTERNAL_CLB_IDS: + payload = call_clb_api("DescribeTargets", {"LoadBalancerId": load_balancer_id}) + bindings = collect_internal_bindings(list(payload.get("Listeners") or []), service) + if not bindings: + continue + groups.append( + { + "loadBalancerId": load_balancer_id, + "listenerIds": sorted({binding["listenerId"] for binding in bindings}), + "bindings": bindings, + } + ) + + if not groups: + return None + + 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), + }, + "groups": groups, + "bindingSummary": render_weight_summary( + [binding for group in groups for binding in group["bindings"]] + ), + "updatedAt": utc_now(), + } + + +def modify_internal_binding_weight(load_balancer_id: str, binding: dict[str, Any], weight: int) -> dict[str, Any]: + # 单条后端单独改权重，出错时能直接定位到具体监听器。 + target: dict[str, Any] = {"Port": int(binding["port"])} + if str(binding.get("instanceId") or "").strip(): + target["InstanceId"] = str(binding["instanceId"]).strip() + else: + # 后端不是按 CVM 实例注册时退回私网 IP 定位。 + target["EniIp"] = str(binding.get("privateIp") or "").strip() + + payload: dict[str, Any] = { + "LoadBalancerId": load_balancer_id, + "ListenerId": binding["listenerId"], + "Targets": [target], + "Weight": int(weight), + } + if str(binding.get("locationId") or "").strip(): + payload["LocationId"] = binding["locationId"] + + response = call_clb_api("ModifyTargetWeight", payload) + wait_clb_api_task(str(response.get("RequestId") or "").strip()) + return { + "routeLabel": binding["routeLabel"], + "weight": int(weight), + "updatedAt": utc_now(), + } + + +def current_internal_bindings(snapshot: dict[str, Any]) -> dict[str, dict[str, Any]]: + # 每次修改后重新读一遍线上值，key 带上 CLB 实例 ID，避免跨 CLB 的绑定互相覆盖。 + current: dict[str, dict[str, Any]] = {} + for group in snapshot["groups"]: + payload = call_clb_api( + "DescribeTargets", + {"LoadBalancerId": group["loadBalancerId"], "ListenerIds": list(group["listenerIds"])}, + ) + for binding in collect_internal_bindings(list(payload.get("Listeners") or []), snapshot["service"]): + current[f"{group['loadBalancerId']}|{binding_key(binding)}"] = binding + return current + + +def wait_internal_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: dict[str, dict[str, Any]] = {} + while time.time() < deadline: + last_bindings = current_internal_bindings(snapshot) + if last_bindings and all( + key in last_bindings and int(last_bindings[key].get("weight") or 0) == int(expected_weights[key]) + for key in expected_weights + ): + return list(last_bindings.values()) + time.sleep(max(float(GATEWAY_CLB_POLL_SECONDS), 0.1)) + + raise RuntimeError( + "internal CLB weight not ready: " + + ", ".join( + f"{last_bindings.get(key, {}).get('routeLabel', key)}={int(last_bindings.get(key, {}).get('weight') or -1)}" + for key in expected_weights + ) + ) + + +def drain_internal_clb_target(snapshot: dict[str, Any]) -> dict[str, Any]: + # 把当前实例在所有内网 CLB 上的权重调成 0，确认生效后再等存量连接排空。 + expected_weights: dict[str, int] = {} + operations: list[dict[str, Any]] = [] + for group in snapshot["groups"]: + for binding in group["bindings"]: + operations.append(modify_internal_binding_weight(group["loadBalancerId"], binding, 0)) + expected_weights[f"{group['loadBalancerId']}|{binding_key(binding)}"] = 0 + + verified = wait_internal_binding_weights( + snapshot, + expected_weights, + timeout_seconds=GATEWAY_CLB_TASK_TIMEOUT_SECONDS, + ) + if INTERNAL_CLB_DRAIN_SECONDS > 0: + time.sleep(INTERNAL_CLB_DRAIN_SECONDS) + return { + "ok": True, + "updatedAt": utc_now(), + "operations": operations, + "drainSeconds": INTERNAL_CLB_DRAIN_SECONDS, + "summary": render_weight_summary(verified), + } + + +def restore_internal_clb_target(snapshot: dict[str, Any]) -> dict[str, Any]: + # 服务恢复后按摘流前的原权重逐条恢复，避免线上权重分配被改坏。 + expected_weights: dict[str, int] = {} + operations: list[dict[str, Any]] = [] + for group in snapshot["groups"]: + for binding in group["bindings"]: + original_weight = int(binding.get("weight") or 0) + operations.append(modify_internal_binding_weight(group["loadBalancerId"], binding, original_weight)) + expected_weights[f"{group['loadBalancerId']}|{binding_key(binding)}"] = original_weight + + verified = wait_internal_binding_weights( + snapshot, + expected_weights, + timeout_seconds=GATEWAY_CLB_TASK_TIMEOUT_SECONDS, + ) + return { + "ok": True, + "updatedAt": utc_now(), + "operations": operations, + "summary": render_weight_summary(verified), + } + + +def collect_internal_health_states(load_balancers: list[dict[str, Any]], snapshot: dict[str, Any]) -> list[dict[str, Any]]: + # DescribeTargetHealth 对 L4 监听器同样返回 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() + rules = list(listener.get("Rules") or []) + for rule in rules: + for target in rule.get("Targets") or []: + if not target_matches_service(target, snapshot["service"]): + continue + state = { + "listenerId": listener_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[f"{listener_id}|{state['port']}"] = state + return list(matched.values()) + + +def wait_internal_clb_target_healthy(snapshot: dict[str, Any], *, timeout_seconds: int) -> list[dict[str, Any]]: + # 等 CLB 自己判定当前实例健康后再恢复权重并继续下一台，避免两台同时不可用。 + expected_keys = { + f"{binding['listenerId']}|{int(binding.get('port') or 0)}" + for group in snapshot["groups"] + for binding in group["bindings"] + } + deadline = time.time() + max(float(timeout_seconds), 1.0) + last_states: list[dict[str, Any]] = [] + while time.time() < deadline: + last_states = [] + for group in snapshot["groups"]: + payload = call_clb_api( + "DescribeTargetHealth", + {"LoadBalancerIds": [group["loadBalancerId"]]}, + ) + last_states.extend( + collect_internal_health_states(list(payload.get("LoadBalancers") or []), snapshot) + ) + + current_by_key = {f"{item['listenerId']}|{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('detail') or item.get('healthStatus')}" + for item in last_states + ) + raise RuntimeError(f"internal CLB health not ready: {detail or 'no matching target health'}") diff --git a/tests/test_internal_clb_drain.py b/tests/test_internal_clb_drain.py new file mode 100644 index 0000000..2b30d5f --- /dev/null +++ b/tests/test_internal_clb_drain.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +import unittest +from unittest.mock import patch + +from monitors.yumi.tencent_clb import ( + collect_internal_bindings, + collect_internal_health_states, + prepare_internal_clb_snapshot, +) + + +APP1_OTHER = { + "host": "app-1", + "service": "other", + "ip": "10.2.11.3", + "instanceId": "ins-n566mmeg", + "port": 2400, +} + +L4_LISTENERS = [ + { + "ListenerId": "lbl-other", + "Protocol": "TCP", + "Port": 2400, + "Targets": [ + {"InstanceId": "ins-n566mmeg", "Port": 2400, "Weight": 10, "PrivateIpAddresses": ["10.2.11.3"]}, + {"InstanceId": "ins-kauk17la", "Port": 2400, "Weight": 10, "PrivateIpAddresses": ["10.2.12.6"]}, + ], + }, + { + "ListenerId": "lbl-live", + "Protocol": "TCP", + "Port": 2500, + "Targets": [ + {"InstanceId": "ins-n566mmeg", "Port": 2500, "Weight": 10, "PrivateIpAddresses": ["10.2.11.3"]}, + ], + }, +] + + +class InternalClbBindingTests(unittest.TestCase): + def test_collect_internal_bindings_matches_only_service_port_and_host(self) -> None: + # 同一台宿主机同时挂 other-2400 / live-2500，只允许命中当前服务自己的监听器。 + bindings = collect_internal_bindings(L4_LISTENERS, APP1_OTHER) + + self.assertEqual(len(bindings), 1) + self.assertEqual(bindings[0]["listenerId"], "lbl-other") + self.assertEqual(bindings[0]["port"], 2400) + self.assertEqual(bindings[0]["weight"], 10) + self.assertEqual(bindings[0]["instanceId"], "ins-n566mmeg") + + @patch("monitors.yumi.tencent_clb.INTERNAL_CLB_IDS", ()) + def test_prepare_snapshot_disabled_without_config(self) -> None: + # 未配置 INTERNAL_CLB_IDS 时必须整体跳过，保持历史行为。 + self.assertIsNone(prepare_internal_clb_snapshot(APP1_OTHER)) + + @patch("monitors.yumi.tencent_clb.call_clb_api") + @patch("monitors.yumi.tencent_clb.INTERNAL_CLB_IDS", ("lb-app", "lb-pay")) + def test_prepare_snapshot_collects_matching_clbs_only(self, call_clb_api_mock) -> None: + def fake_describe(method: str, payload: dict) -> dict: + self.assertEqual(method, "DescribeTargets") + if payload["LoadBalancerId"] == "lb-app": + return {"Listeners": L4_LISTENERS} + return {"Listeners": []} + + call_clb_api_mock.side_effect = fake_describe + + snapshot = prepare_internal_clb_snapshot(APP1_OTHER) + + self.assertIsNotNone(snapshot) + self.assertEqual(len(snapshot["groups"]), 1) + group = snapshot["groups"][0] + self.assertEqual(group["loadBalancerId"], "lb-app") + self.assertEqual(group["listenerIds"], ["lbl-other"]) + self.assertEqual(len(group["bindings"]), 1) + self.assertEqual(int(group["bindings"][0]["weight"]), 10) + + @patch("monitors.yumi.tencent_clb.call_clb_api") + @patch("monitors.yumi.tencent_clb.INTERNAL_CLB_IDS", ("lb-app",)) + def test_prepare_snapshot_none_when_service_not_behind_clb(self, call_clb_api_mock) -> None: + call_clb_api_mock.return_value = {"Listeners": []} + + self.assertIsNone(prepare_internal_clb_snapshot(APP1_OTHER)) + + +class InternalClbHealthTests(unittest.TestCase): + def test_collect_internal_health_states_flattens_l4_rules(self) -> None: + load_balancers = [ + { + "LoadBalancerId": "lb-app", + "Listeners": [ + { + "ListenerId": "lbl-other", + "Rules": [ + { + "Targets": [ + { + "TargetId": "ins-n566mmeg", + "Port": 2400, + "HealthStatus": True, + "HealthStatusDetail": "Alive", + }, + { + "TargetId": "ins-kauk17la", + "Port": 2400, + "HealthStatus": False, + "HealthStatusDetail": "Dead", + }, + ] + } + ], + } + ], + } + ] + snapshot = {"service": APP1_OTHER} + + states = collect_internal_health_states(load_balancers, snapshot) + + self.assertEqual(len(states), 1) + self.assertEqual(states[0]["targetId"], "ins-n566mmeg") + self.assertTrue(states[0]["healthStatus"]) + + +if __name__ == "__main__": + unittest.main()