From 649f348540a5c54c962818fe2b9511959285fbcf Mon Sep 17 00:00:00 2001 From: 170-carry Date: Mon, 27 Apr 2026 23:35:31 +0800 Subject: [PATCH] Fix nacos drain state detection --- monitors/yumi/service_ops.py | 32 ++++++++++++- tests/test_nacos_drain_state.py | 80 +++++++++++++++++++++++++++++++++ 2 files changed, 110 insertions(+), 2 deletions(-) create mode 100644 tests/test_nacos_drain_state.py diff --git a/monitors/yumi/service_ops.py b/monitors/yumi/service_ops.py index 297277d..958a0d5 100644 --- a/monitors/yumi/service_ops.py +++ b/monitors/yumi/service_ops.py @@ -235,6 +235,7 @@ def wait_java_registration_enabled_state(service: dict[str, Any], enabled: bool, namespace_id, _ = current_namespace() deadline = time.time() + timeout_seconds last_record: dict[str, Any] | None = None + last_source = "" while time.time() < deadline: last_record = matched_nacos_record(list_instances(namespace_id, registry_name, NACOS_DISCOVERY_GROUP), service) @@ -245,9 +246,31 @@ def wait_java_registration_enabled_state(service: dict[str, Any], enabled: bool, "healthy": nacos_truthy(last_record.get("healthy", False)), "source": "instance/list", } + if last_record is not None: + last_source = "instance/list" + + # Nacos 的 instance/list 可能不返回 disabled 实例，catalog/instances 仍能看到完整状态。 + catalog_record = matched_nacos_record(catalog_instances(namespace_id, registry_name, NACOS_DISCOVERY_GROUP), service) + if catalog_record is not None and nacos_truthy(catalog_record.get("enabled", True)) == enabled: + return { + "checked": True, + "enabled": enabled, + "healthy": nacos_truthy(catalog_record.get("healthy", False)), + "source": "catalog/instances", + } + if catalog_record is not None: + last_record = catalog_record + last_source = "catalog/instances" + time.sleep(ROLLING_RESTART_POLL_SECONDS) - raise RuntimeError(f"nacos enabled={enabled} not ready on {service['host']} {service['service']}") + detail = "" + if last_record is not None: + detail = ( + f" last {last_source or 'nacos'} enabled={last_record.get('enabled')} " + f"healthy={last_record.get('healthy')} ip={last_record.get('ip')} port={last_record.get('port')}" + ) + raise RuntimeError(f"nacos enabled={enabled} not ready on {service['host']} {service['service']}{detail}") def drain_java_registration(service: dict[str, Any]) -> dict[str, Any]: @@ -256,7 +279,12 @@ def drain_java_registration(service: dict[str, Any]) -> dict[str, Any]: return {"checked": False, "reason": "service not configured for nacos release drain"} update_result = update_java_registration_enabled(service, False) - state_result = wait_java_registration_enabled_state(service, False, NACOS_RELEASE_DRAIN_TIMEOUT_SECONDS) + try: + state_result = wait_java_registration_enabled_state(service, False, NACOS_RELEASE_DRAIN_TIMEOUT_SECONDS) + except Exception: + # 如果摘流确认失败，尽量立刻恢复 enabled=true，避免业务实例长期停留在 Nacos 禁用状态。 + update_java_registration_enabled(service, True) + raise if NACOS_RELEASE_DRAIN_SECONDS > 0: time.sleep(NACOS_RELEASE_DRAIN_SECONDS) return { diff --git a/tests/test_nacos_drain_state.py b/tests/test_nacos_drain_state.py new file mode 100644 index 0000000..2b0b146 --- /dev/null +++ b/tests/test_nacos_drain_state.py @@ -0,0 +1,80 @@ +from __future__ import annotations + +import unittest +from unittest.mock import call, patch + +from monitors.yumi import service_ops + + +class NacosDrainStateTests(unittest.TestCase): + def test_enabled_state_uses_catalog_when_disabled_instance_is_filtered_from_instance_list(self) -> None: + service = { + "host": "app-1", + "service": "other", + "ip": "10.2.11.3", + "port": 2400, + "kind": "java", + } + + with ( + patch.object(service_ops, "current_namespace", return_value=("namespace-id", "namespace")), + patch.object( + service_ops, + "list_instances", + return_value=[ + { + "ip": "10.2.12.6", + "port": 2400, + "healthy": True, + "enabled": True, + } + ], + ), + patch.object( + service_ops, + "catalog_instances", + return_value=[ + { + "ip": "10.2.11.3", + "port": 2400, + "healthy": True, + "enabled": False, + } + ], + ), + ): + result = service_ops.wait_java_registration_enabled_state(service, False, 1) + + self.assertTrue(result["checked"]) + self.assertFalse(result["enabled"]) + self.assertTrue(result["healthy"]) + self.assertEqual(result["source"], "catalog/instances") + + def test_drain_restores_enabled_when_wait_state_fails_after_update(self) -> None: + service = { + "host": "app-1", + "service": "other", + "ip": "10.2.11.3", + "port": 2400, + "kind": "java", + } + + with ( + patch.object(service_ops, "should_drain_java_registration", return_value=True), + patch.object(service_ops, "wait_java_registration_enabled_state", side_effect=RuntimeError("not ready")), + patch.object(service_ops, "update_java_registration_enabled", return_value={"checked": True}) as update_mock, + ): + with self.assertRaises(RuntimeError): + service_ops.drain_java_registration(service) + + self.assertEqual( + update_mock.call_args_list, + [ + call(service, False), + call(service, True), + ], + ) + + +if __name__ == "__main__": + unittest.main()