Fix nacos drain state detection

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 {

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