From ab6791cf342c1164864183c61ee4d462d47cfdb7 Mon Sep 17 00:00:00 2001 From: lili Date: Wed, 1 Jul 2026 11:47:22 -0700 Subject: [PATCH] =?UTF-8?q?feat(gen-worker):=20AgentScope=20Service=20?= =?UTF-8?q?=E6=B3=A8=E5=86=8C=E8=BF=9B=20Nacos=20discovery(=E6=9B=BF?= =?UTF-8?q?=E7=A1=AC=E7=BC=96=E7=A0=81@8200)(=E5=88=87=E7=89=87=E4=B8=80?= =?UTF-8?q?=20=E9=98=B6=E6=AE=B5=E3=80=87=20D1)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit nacos_registry 注册实例+心跳+优雅注销,advertise_ip 指 mini-desktop Tailscale;代理旁路; NACOS_REGISTER=1 才开、best-effort 失败不中断 Service。单测 mock client 锁注册+旁路+best-effort。 Co-Authored-By: Claude Opus 4.8 (1M context) --- tier2/gen-worker/requirements.txt | 7 + tier2/gen-worker/service/app.py | 9 ++ tier2/gen-worker/service/nacos_registry.py | 108 +++++++++++++ tier2/gen-worker/tests/test_nacos_registry.py | 149 ++++++++++++++++++ 4 files changed, 273 insertions(+) create mode 100644 tier2/gen-worker/service/nacos_registry.py create mode 100644 tier2/gen-worker/tests/test_nacos_registry.py diff --git a/tier2/gen-worker/requirements.txt b/tier2/gen-worker/requirements.txt index 9beba7ec..bd76fdc5 100644 --- a/tier2/gen-worker/requirements.txt +++ b/tier2/gen-worker/requirements.txt @@ -12,5 +12,12 @@ pyyaml # 模型路由 models.yaml + infra.yaml 基建配置 pymysql # MySQL 源工程版本 manifest 表（tier2_source_project_version） minio # MinIO（S3 兼容 OSS）源文件全文落库 +# 服务发现（配置控制面 §3.7 Part D · D1）：Agent Service 注册进 Nacos discovery，让 Java 按服务名发现， +# 替硬编码 @8200。钉 v1（<2）—— v1 的同步 nacos.NacosClient（add/remove_naming_instance + 自带守护心跳 +# 线程，免 asyncio event-loop）；与 agentscope==2.0.2 同「钉版本」纪律，锁 1.0.0 保可复现。仅 +# service/nacos_registry.py 惰性 import，且 NACOS_REGISTER=1 才用；默认关 / 本地 / 6c6g 不装也能 +# py_compile + import service.app。 +nacos-sdk-python==1.0.0 + # 注：M3 走 AgentScope 内置 AnthropicChatModel（已含在 agentscope）。 # 注：Phaser / esbuild 是 JS 侧工具链（game-runtime / tier2/harness），不在此 Python 依赖内。 diff --git a/tier2/gen-worker/service/app.py b/tier2/gen-worker/service/app.py index 6e3a4559..a80f3ac2 100644 --- a/tier2/gen-worker/service/app.py +++ b/tier2/gen-worker/service/app.py @@ -274,6 +274,15 @@ def main() -> None: reload = os.environ.get("TIER2_SERVICE_RELOAD", "0").strip().lower() in ("1", "true", "yes", "on") print(f"[tier2-service] 启动 Agent Service:http://{host}:{port} reload={reload}", flush=True) + + # Nacos discovery 注册(配置控制面 §3.7 Part D):NACOS_REGISTER=1 才注册,默认关 —— 不影响本地 / CLI 跑 + # (也不牵 nacos 依赖)。register 是 best-effort:注册失败只告警、不抛,不中断本 Service 起服务。 + # 注意:注册进 Nacos 的通告 IP 是 SERVICE_ADVERTISE_IP(mini-desktop Tailscale 100.64.0.7),不是绑定 + # host;host 只作日志区分。reload 调试模式一般不开 NACOS_REGISTER,此处置于两分支之前保持一处收口。 + if os.environ.get("NACOS_REGISTER") == "1": + from . import nacos_registry # noqa: PLC0415 —— 惰性 import,默认关时不牵 nacos 依赖 + nacos_registry.register(host, port) + if reload: # reload 模式必须传 import 字符串(uvicorn 要能在 worker 子进程重导入);指向本模块的 module-level `app`。 # 子进程重导入 service.app 时,module-level `app` 只在 TIER2_SERVICE_EAGER_APP=1 时才建——故在此 diff --git a/tier2/gen-worker/service/nacos_registry.py b/tier2/gen-worker/service/nacos_registry.py new file mode 100644 index 00000000..b6d25758 --- /dev/null +++ b/tier2/gen-worker/service/nacos_registry.py @@ -0,0 +1,108 @@ +"""nacos_registry.py —— AgentScope Service 注册进 Nacos discovery(替硬编码 @8200,让 Java 消费方按服务名发现)。 + +配置控制面设计 §3.7 Part D:生产形态是「Java 管准入+队列(Part C),Python 只服务 /chat」。本模块让 +tier2 Agent Service(跑在 mini-desktop)把自己注册进 Nacos naming(服务发现),Java 侧按 `service-name` +发现它,不再硬编码 `100.64.0.7:8200`。 + +设计要点(逐条对着 nacos-sdk-python v1.0.0 的实际行为写,已核验源码): + 1. 用 v1 的同步 `nacos.NacosClient`(含 `add_naming_instance` 注册 / `remove_naming_instance` 注销), + 不牵 asyncio event-loop(与 AgentScope Service 的 uvicorn 事件循环互不打架)。 + 2. **代理旁路(内网铁律)**:本机系统代理是 fake-ip `198.18.x`,nacos client 走 HTTP 内网直连 + `100.64.0.8:8848` 必须绕过它 —— 把 Nacos host 塞进 `NO_PROXY`/`no_proxy`,**且必须在建 client 前** + 设好(urllib 建连时读环境;晚设不生效,内网直连会被 fake-ip 代理拦、栽 502/连接关闭)。 + 3. **ephemeral + 心跳**:`add_naming_instance(..., ephemeral=True)` 注册的是临时实例,Nacos 若收不到心跳 + 会在健康超时(默认 ~15s)后自动摘除。故必须传 `heartbeat_interval`,让 SDK 起一个**守护线程** + (`HeartbeatTask.daemon=True`,核验 nacos/task.py)周期性 `send_heartbeat` 续命 —— 这样实例才真正 + 「活着可被发现」。守护线程不阻塞进程退出;进程崩溃时心跳随之停,Nacos 到期自动摘除(临时实例自愈, + 比 persistent 实例「进程死了还赖在注册表」更干净)。brief 初版漏传 heartbeat_interval,会注册后 ~15s + 即消失、达不到「让 Java 发现」的目的,此处按「生产稳定性 > 开发期简单」补上。 + 4. **优雅注销**:`atexit` 注册退出钩子,进程正常退出时 `remove_naming_instance`(v1 源码里会一并 + `stop()` 掉那条心跳守护线程),避免 Nacos 里留死实例。 + 5. **best-effort**:注册全过程包在 try/except 里 —— 注册失败(Nacos 不可达 / 鉴权失败等)只告警、不抛、 + 不中断 Service 起。生成主链(/chat)不依赖「被发现」,发现只为 Java 消费方定位;发现挂了不该拖垮生成。 + +调用方:`service/app.py` 的 `main()` 在 `uvicorn.run` 之前、`NACOS_REGISTER=1` 时调 `register(host, port)`。 +默认关(不设 NACOS_REGISTER),本地 / CLI 跑不受影响、不牵 nacos 依赖。 +""" + +from __future__ import annotations + +import atexit +import os + +# 注册用的服务名(Java 侧按这个名字发现);与设计 §3.7 Part D 约定一致。 +_SERVICE_NAME = "agentscope-gen-service" + + +def _bypass_proxy_for(host: str) -> None: + """把 host 并进 NO_PROXY/no_proxy(内网直连绕系统 fake-ip 代理;必须在建 nacos client 前调)。 + + NO_PROXY / no_proxy 两个大小写变体都写,兼容不同 HTTP 栈的读取习惯;已在列表里就不重复追加。 + """ + for key in ("NO_PROXY", "no_proxy"): + cur = os.environ.get(key, "") + # 逗号分隔、去空白后精确比对各段,避免 "100.64.0.8" 与 "100.64.0.80" 之类子串误判。 + segments = [seg.strip() for seg in cur.split(",") if seg.strip()] + if host not in segments: + segments.append(host) + os.environ[key] = ",".join(segments) + + +def register(service_host: str, service_port: int) -> None: + """把本 Service 实例注册进 Nacos naming(best-effort;失败告警不抛,不中断 Service 起)。 + + 参数: + service_host —— Service 的**绑定** host(app.py 里的 TIER2_SERVICE_HOST,默认 0.0.0.0)。注意: + 注册进 Nacos 的 IP **不是**它(0.0.0.0 不可路由),而是下面 `advertise_ip` —— Java 侧 + 能连到的地址(mini-desktop Tailscale 100.64.0.7)。此参数仅用于日志里显式区分 + 「绑定 host」与「对外通告 IP」,避免误解。 + service_port —— Service 端口(TIER2_SERVICE_PORT,默认 8200),注册进 Nacos 的就是它。 + """ + nacos_addr = os.environ.get("NACOS_SERVER_ADDR", "100.64.0.8:8848") + # 内网铁律:必须在建 client 前把 Nacos host 塞进 NO_PROXY,否则 HTTP 直连被 fake-ip 代理拦。 + _bypass_proxy_for(nacos_addr.split(":")[0]) + + advertise_ip = os.environ.get("SERVICE_ADVERTISE_IP", "100.64.0.7") + # 心跳周期(秒):>0 才让 SDK 起守护线程续命临时实例;设 0 则退化为不续命(实例 ~15s 后被摘,一般不用)。 + try: + heartbeat_interval = int(os.environ.get("NACOS_HEARTBEAT_INTERVAL", "5")) + except ValueError: + heartbeat_interval = 5 + + try: + import nacos # noqa: PLC0415 —— 惰性 import:默认不开注册时不牵 nacos 依赖(本地/CLI/6c6g 免装) + + client = nacos.NacosClient( + nacos_addr, + namespace=os.environ.get("NACOS_NAMESPACE", "public"), + username=os.environ.get("NACOS_USERNAME", "nacos"), + password=os.environ.get("NACOS_PASSWORD", ""), + ) + # ip 用「Java 侧能连到的」通告地址(不是绑定 host 0.0.0.0);heartbeat_interval 让 SDK 起守护线程续命。 + client.add_naming_instance( + _SERVICE_NAME, + advertise_ip, + service_port, + healthy=True, + heartbeat_interval=heartbeat_interval if heartbeat_interval > 0 else None, + ) + print( + f"[nacos-registry] 已注册 {_SERVICE_NAME} @ {advertise_ip}:{service_port} " + f"(bind_host={service_host},nacos={nacos_addr},heartbeat={heartbeat_interval}s)", + flush=True, + ) + + def _deregister() -> None: + """进程退出钩子:摘掉本实例并停掉心跳守护线程(v1 remove_naming_instance 内部会 stop 心跳)。""" + try: + client.remove_naming_instance(_SERVICE_NAME, advertise_ip, service_port) + print(f"[nacos-registry] 已注销 {_SERVICE_NAME} @ {advertise_ip}:{service_port}", flush=True) + except Exception as e: # noqa: BLE001 —— 注销失败只忽略,别在退出路径上抛 + print(f"[nacos-registry] 注销失败(忽略):{type(e).__name__}: {e}", flush=True) + + atexit.register(_deregister) + except Exception as e: # noqa: BLE001 —— best-effort:注册失败不连累 Service 起 + print( + f"[nacos-registry] 注册失败(best-effort,不中断 Service):{type(e).__name__}: {e}", + flush=True, + ) diff --git a/tier2/gen-worker/tests/test_nacos_registry.py b/tier2/gen-worker/tests/test_nacos_registry.py new file mode 100644 index 00000000..0ec46676 --- /dev/null +++ b/tier2/gen-worker/tests/test_nacos_registry.py @@ -0,0 +1,149 @@ +"""test_nacos_registry.py —— D1 Nacos 注册封装单测(mock nacos.NacosClient,不触网)。 + +配置控制面 §3.7 Part D · D1。验四件事(brief Step 4 的三条 + 一条回归护栏): + 1. register 以 (service_name, advertise_ip, port) 调 add_naming_instance —— 通告 IP 是 100.64.0.7, + 不是绑定 host 0.0.0.0。 + 2. register 后 NO_PROXY / no_proxy 都含 Nacos host(fake-ip 代理旁路铁律),且旁路在【建 client 前】完成。 + 3. NacosClient 构造抛异常时 register **不抛**(best-effort),也不挂注销钩子。 + 4. (回归护栏)注册传了非 None 的 heartbeat_interval —— 临时实例靠 SDK 守护心跳续命,漏传会 ~15s 被 Nacos 摘。 + 另附:注销钩子触发时调 remove_naming_instance 摘实例。 + +全程 mock nacos.NacosClient,不连真 Nacos、不触网。跑法: + PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/test_nacos_registry.py -v +""" + +from __future__ import annotations + +import os +import sys +from unittest import mock + +import pytest + +# 包内/直跑兼容:把 gen-worker/ 加进 sys.path(PYTHONPATH 已指 gen-worker 时这步是冗余保险)。 +_HERE = os.path.dirname(os.path.abspath(__file__)) +_GW = os.path.dirname(_HERE) # gen-worker/ +if _GW not in sys.path: + sys.path.insert(0, _GW) + +from service import nacos_registry # noqa: E402 —— 被测模块 + +_NACOS_HOST = "100.64.0.8" +_ADVERTISE_IP = "100.64.0.7" + + +@pytest.fixture() +def clean_env(): + """隔离 env:测前快照、测后还原。 + + register 会**直接改** os.environ[NO_PROXY](非经 monkeypatch),故手动快照 + finally 还原, + 防止跨测 / 跨会话泄漏。同时固定 Nacos 地址与通告 IP,让断言可预期。 + """ + keys = ( + "NO_PROXY", "no_proxy", "NACOS_SERVER_ADDR", "NACOS_NAMESPACE", + "NACOS_USERNAME", "NACOS_PASSWORD", "NACOS_REGISTER", + "SERVICE_ADVERTISE_IP", "NACOS_HEARTBEAT_INTERVAL", + ) + snapshot = {k: os.environ.get(k) for k in keys} + for k in ("NO_PROXY", "no_proxy", "NACOS_REGISTER", "NACOS_HEARTBEAT_INTERVAL"): + os.environ.pop(k, None) + os.environ["NACOS_SERVER_ADDR"] = f"{_NACOS_HOST}:8848" + os.environ["SERVICE_ADVERTISE_IP"] = _ADVERTISE_IP + try: + yield + finally: + for k, v in snapshot.items(): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v + + +def test_register_calls_add_naming_instance(clean_env): + """register 应以 (service_name, advertise_ip, port) 调 add_naming_instance,且带非 None 心跳。""" + fake_client = mock.MagicMock() + with mock.patch("nacos.NacosClient", return_value=fake_client) as ctor, \ + mock.patch.object(nacos_registry.atexit, "register") as atexit_reg: + nacos_registry.register("0.0.0.0", 8200) + + # 构造 client:server_addr 位置参 + namespace/username/password 关键字参 + ctor.assert_called_once() + c_args, c_kwargs = ctor.call_args + assert c_args[0] == f"{_NACOS_HOST}:8848" + assert c_kwargs.get("namespace") == "public" + assert c_kwargs.get("username") == "nacos" + + # add_naming_instance:通告 IP(非绑定 0.0.0.0)+ port + fake_client.add_naming_instance.assert_called_once() + a_args, a_kwargs = fake_client.add_naming_instance.call_args + assert a_args[0] == "agentscope-gen-service" + assert a_args[1] == _ADVERTISE_IP # 通告 IP,不是绑定 host 0.0.0.0 + assert a_args[2] == 8200 + assert a_kwargs.get("healthy") is True + # 回归护栏:必须传非 None 且 >0 的心跳,否则临时实例 ~15s 后被 Nacos 摘除 + assert a_kwargs.get("heartbeat_interval") is not None + assert a_kwargs["heartbeat_interval"] > 0 + + # 注册成功后应挂优雅注销钩子 + atexit_reg.assert_called_once() + + +def test_register_bypasses_proxy(clean_env): + """register 后 NO_PROXY / no_proxy 都应含 Nacos host(fake-ip 代理旁路)。""" + fake_client = mock.MagicMock() + with mock.patch("nacos.NacosClient", return_value=fake_client), \ + mock.patch.object(nacos_registry.atexit, "register"): + nacos_registry.register("0.0.0.0", 8200) + + assert _NACOS_HOST in os.environ.get("NO_PROXY", "") + assert _NACOS_HOST in os.environ.get("no_proxy", "") + + +def test_proxy_bypass_happens_before_client_build(clean_env): + """内网铁律:代理旁路必须在【建 client 前】完成 —— 构造 NacosClient 时 NO_PROXY 已含 host。""" + seen = {} + + def _spy_ctor(*_args, **_kwargs): + seen["no_proxy_at_build"] = os.environ.get("NO_PROXY", "") + return mock.MagicMock() + + with mock.patch("nacos.NacosClient", side_effect=_spy_ctor), \ + mock.patch.object(nacos_registry.atexit, "register"): + nacos_registry.register("0.0.0.0", 8200) + + assert _NACOS_HOST in seen.get("no_proxy_at_build", "") + + +def test_register_best_effort_on_ctor_error(clean_env): + """NacosClient 构造抛异常时 register 不抛(best-effort),且不挂注销钩子。""" + with mock.patch("nacos.NacosClient", side_effect=RuntimeError("nacos unreachable")), \ + mock.patch.object(nacos_registry.atexit, "register") as atexit_reg: + nacos_registry.register("0.0.0.0", 8200) # 不应抛异常 + atexit_reg.assert_not_called() + + +def test_register_best_effort_on_add_instance_error(clean_env): + """add_naming_instance 抛异常时 register 也不抛(best-effort 覆盖到注册请求失败)。""" + fake_client = mock.MagicMock() + fake_client.add_naming_instance.side_effect = RuntimeError("register 403") + with mock.patch("nacos.NacosClient", return_value=fake_client), \ + mock.patch.object(nacos_registry.atexit, "register") as atexit_reg: + nacos_registry.register("0.0.0.0", 8200) # 不应抛异常 + atexit_reg.assert_not_called() + + +def test_deregister_removes_instance(clean_env): + """atexit 注销钩子触发时应以 (service_name, advertise_ip, port) 调 remove_naming_instance。""" + fake_client = mock.MagicMock() + captured = {} + + with mock.patch("nacos.NacosClient", return_value=fake_client), \ + mock.patch.object( + nacos_registry.atexit, "register", + side_effect=lambda fn: captured.__setitem__("deregister", fn)): + nacos_registry.register("0.0.0.0", 8200) + + captured["deregister"]() # 手动触发退出钩子 + fake_client.remove_naming_instance.assert_called_once_with( + "agentscope-gen-service", _ADVERTISE_IP, 8200 + )