feat(gen-worker): AgentScope Service 注册进 Nacos discovery(替硬编码@8200)(切片一 阶段〇 D1)
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) <noreply@anthropic.com>
This commit is contained in:
lili
parent
4bc36ec6dd
commit
ab6791cf34
@ -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 依赖内。
|
||||
|
||||
@ -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 时才建——故在此
|
||||
|
||||
108
tier2/gen-worker/service/nacos_registry.py
Normal file
108
tier2/gen-worker/service/nacos_registry.py
Normal file
@ -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,
|
||||
)
|
||||
149
tier2/gen-worker/tests/test_nacos_registry.py
Normal file
149
tier2/gen-worker/tests/test_nacos_registry.py
Normal file
@ -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
|
||||
)
|
||||
Loading…
x
Reference in New Issue
Block a user