hy/hy-app-monitor
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: drain internal CLB targets before restarting service instances

Browse Source 
部署/滚动重启非 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 <noreply@anthropic.com>
This commit is contained in:
zhx 2026-07-10 12:56:27 +08:00
parent c493d335e6
commit f89ebe797e
6 changed files with 541 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.env.example
View File

@ -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

7
core/config.py
View File

@ -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:

37
monitors/yumi/service_ops.py
View File

@ -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)

57
monitors/yumi/service_release.py
View File

@ -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)

311
monitors/yumi/tencent_clb.py
View File

@ -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'}")

127
tests/test_internal_clb_drain.py Normal file
View File

@ -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()