diff --git a/tier2/gen-worker/scripts/demo_nacos_hot_read.py b/tier2/gen-worker/scripts/demo_nacos_hot_read.py new file mode 100644 index 00000000..a43cce30 --- /dev/null +++ b/tier2/gen-worker/scripts/demo_nacos_hot_read.py @@ -0,0 +1,147 @@ +#!/usr/bin/env python3 +"""demo_nacos_hot_read.py —— 真 Nacos 热读端到端演示(配置控制面阶段二·步骤4 证据脚本)。 + +【这个脚本证明什么】 +在一个进程内同时扮演「写侧」(模拟 yudao 激活编排 publish 生效 dataId)与「读侧」(worker 起本工厂 +接 Nacos 热源),坐实设计 §4 步骤4「验证」头一条:改 Nacos 生效 dataId → worker 数秒内经 genconfig +读到新预算/阈值,不重启。流程: + 1. 起读侧工厂(启用旗打开 + 指向【演示专用 dataId】,绝不碰生产 gen-hot-params-tier2/cheap); + 2. 写侧 publish {budget.rmb_hard_limit: 33.0} → 断言 genconfig 数秒内热读到 33.0; + 3. 改值 publish 44.0 → 断言再热读到 44.0; + 4. 清理:删除演示 dataId、detach 读侧、验证回落本地。 + +【不动 live】只碰演示专用 dataId(gen-hot-params-demo),不部署、不重启任何 live 进程。 + +【内网铁律】Nacos 走 HTTP 直连内网 100.64.0.8:8848,必须在建 client 前把 host 塞进 NO_PROXY 绕系统 +fake-ip 代理(工厂内已做;本脚本的独立 publish client 也自旁路一次)。凭据从 env 读(NACOS_PASSWORD), +权威副本在 docs/内网凭据与端点.md;脚本不硬编码口令。 + +跑法(在 gen-worker/ 下,主仓解释器): + NACOS_PASSWORD=<读凭据档> PYTHONPATH=tier2/gen-worker \ + cheap-worker/.venv/bin/python tier2/gen-worker/scripts/demo_nacos_hot_read.py +""" + +from __future__ import annotations + +import json +import os +import sys +import time + +# 包内/直跑兼容:把 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) + +# 演示专用 dataId(绝不碰生产 gen-hot-params-tier2 / gen-hot-params-cheap)。 +DEMO_DATA_ID = "gen-hot-params-demo" +GROUP = "DEFAULT_GROUP" +AREA, KEY = "budget", "rmb_hard_limit" +COMPOSITE = f"{AREA}.{KEY}" # genconfig 热源复合键口径 + + +def _bypass_proxy(host: str) -> None: + """把 host 并进 NO_PROXY/no_proxy(建 publish client 前;与工厂/nacos_registry 同款内网旁路)。""" + for k in ("NO_PROXY", "no_proxy"): + cur = os.environ.get(k, "") + segs = [s.strip() for s in cur.split(",") if s.strip()] + if host not in segs: + segs.append(host) + os.environ[k] = ",".join(segs) + + +def _wait_until(pred, timeout: float = 20.0, interval: float = 0.5): + """轮询等 pred() 为真;返回耗时秒,超时返 None。用于等长轮询推送到达。""" + start = time.time() + while time.time() - start < timeout: + if pred(): + return time.time() - start + time.sleep(interval) + return None + + +def _read() -> float: + """当前 genconfig 读到的 budget.rmb_hard_limit(经热源→env→YAML→内置默认取值链)。""" + from worker import genconfig + return genconfig.get(AREA, KEY, 50.0) + + +def main() -> int: + nacos_addr = os.environ.get("NACOS_SERVER_ADDR", "100.64.0.8:8848") + host = nacos_addr.split(":")[0] + _bypass_proxy(host) + + if not os.environ.get("NACOS_PASSWORD"): + print("[demo] 缺 NACOS_PASSWORD(读 docs/内网凭据与端点.md,export 后重跑)", flush=True) + return 2 + + # 读侧工厂:启用旗打开 + 指向演示专用 dataId(env 覆盖档默认)。 + os.environ["TIER2_GENCONFIG_NACOS"] = "1" + os.environ["TIER2_GENCONFIG_NACOS_DATAID"] = DEMO_DATA_ID + + import nacos # noqa: PLC0415 + from worker import genconfig, genconfig_nacos # noqa: PLC0415 + + # 写侧独立 publish client(模拟 yudao 激活编排的写侧;鉴权 + 内网旁路已就绪)。 + pub = nacos.NacosClient( + nacos_addr, + namespace=os.environ.get("NACOS_NAMESPACE", ""), # 默认公共空间空串(勿写 public) + username=os.environ.get("NACOS_USERNAME", "nacos"), + password=os.environ["NACOS_PASSWORD"], + ) + # 清掉可能的旧演示残留,保证从「空 dataId」起步。 + try: + pub.remove_config(DEMO_DATA_ID, GROUP) + except Exception as e: # noqa: BLE001 + print(f"[demo] 预清理(忽略):{type(e).__name__}: {e}", flush=True) + + baseline = _read() + print(f"[demo] 纯本地基线 {COMPOSITE} = {baseline}(未接热源前的 YAML/内置默认)", flush=True) + + hot = genconfig_nacos.build_and_attach_hot_source(tier="tier2") + if hot is None: + print("[demo] FAIL 工厂未 attach(旗未开或构造失败)", flush=True) + return 1 + print(f"[demo] 读侧已订阅 dataId={DEMO_DATA_ID};演示 dataId 现为空 → genconfig 回落本地 = {_read()}", flush=True) + + # ① publish 33.0 → 断言数秒内热读到。 + v1 = 33.0 + ok1 = pub.publish_config(DEMO_DATA_ID, GROUP, json.dumps({COMPOSITE: v1})) + print(f"[demo] 写侧 publish {COMPOSITE}={v1}(publish_config→{ok1}),等热读推送…", flush=True) + took1 = _wait_until(lambda: _read() == v1) + if took1 is None: + print(f"[demo] FAIL 超时未热读到 {v1}(当前 {_read()})", flush=True) + pub.remove_config(DEMO_DATA_ID, GROUP) + return 1 + print(f"[demo] OK genconfig {took1:.2f}s 内热读到 {COMPOSITE}={_read()}(worker 未重启)", flush=True) + + # ② 改值 44.0 → 再断言热读到。 + v2 = 44.0 + pub.publish_config(DEMO_DATA_ID, GROUP, json.dumps({COMPOSITE: v2})) + print(f"[demo] 写侧改值 publish {COMPOSITE}={v2},等热读推送…", flush=True) + took2 = _wait_until(lambda: _read() == v2) + if took2 is None: + print(f"[demo] FAIL 超时未热读到改后值 {v2}(当前 {_read()})", flush=True) + pub.remove_config(DEMO_DATA_ID, GROUP) + return 1 + print(f"[demo] OK genconfig {took2:.2f}s 内热读到改后值 {COMPOSITE}={_read()}", flush=True) + + # ③ 清理:删演示 dataId + detach 读侧 + 验证回落本地。 + pub.remove_config(DEMO_DATA_ID, GROUP) + genconfig_nacos.detach() + genconfig.reload() + print(f"[demo] 已删除演示 dataId={DEMO_DATA_ID}、detach 读侧;detach 后 {COMPOSITE} 回落本地 = {_read()}", flush=True) + print("[demo] 演示成功:改 Nacos 生效 dataId → genconfig 数秒热读到新值、不重启;清理完毕、未碰生产 dataId。", flush=True) + return 0 + + +if __name__ == "__main__": + _rc = main() + # nacos v1 SDK 的后台守护轮询线程(_do_pulling / _process_polling_result)在进程退出时会因其 + # multiprocessing 队列被关闭抛 EOFError —— 这是 SDK 的 teardown 噪声,与演示结果无关(所有断言在此 + # 之前已打印 OK),且生产 worker 是 uvicorn 长驻进程、不退出、根本不触发。此处 flush 后 os._exit + # 干净终止、跳过该 daemon 线程的 teardown,让证据输出无歧义。 + sys.stdout.flush() + sys.stderr.flush() + os._exit(_rc) diff --git a/tier2/gen-worker/tests/test_genconfig_nacos.py b/tier2/gen-worker/tests/test_genconfig_nacos.py new file mode 100644 index 00000000..f08cc8f3 --- /dev/null +++ b/tier2/gen-worker/tests/test_genconfig_nacos.py @@ -0,0 +1,336 @@ +"""test_genconfig_nacos.py —— genconfig 接 Nacos 热源(配置控制面阶段二·步骤4)单测。 + +全程 mock / fake,不触真 Nacos。锁死设计 §4 步骤4「验证」要求的四条异常/回落路径,外加两档隔离、 +优先级、门阈值热读、脏值兜底与工厂 best-effort: + + 回落链四条(先锁死才许真接线): + ① 未 start / 未 attach → genconfig 走原链(env→YAML→内置默认),行为 = 现状; + 热源已 attach 但缺该键 → 回落 env(热源无值不截断)。 + ② Nacos 不可达 → NacosHotConfig 磁盘快照/默认兜底(get_config 抛)→ store 空 → genconfig 回落本地。 + ③ namespace 配错 → 订阅的空间/dataId 收不到值(get_config 返 None)→ store 空 → 回落本地 + (佐证前置任务:写读两侧 namespaceId 必须同一真实值,否则配置静默推不到)。 + ④ YAML 回落 → 热源无该键、无 env → 取本地 YAML 与内置默认。 + + 另:优先级 Nacos > env(命中即压过 env);cheap 与 tier2 两档各读各 dataId 的热源、不串档; + 软预算(budget)与门阈值(gates)都吃上热读;脏热值保旧回落不穿透;工厂旗默认关 / best-effort / + 按档 dataId / NO_PROXY 在建 client 前 / 回退开关 detach。 + +跑法: + PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest \ + tier2/gen-worker/tests/test_genconfig_nacos.py -v +""" + +from __future__ import annotations + +import json +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 worker import genconfig, genconfig_nacos # noqa: E402 —— 被测模块 +from worker.nacos_hotconfig import NacosHotConfig # noqa: E402 + + +# 所有测试可能碰到的 env,fixture 统一快照 + 还原,防跨测/跨会话泄漏。 +_ENV_KEYS = ( + "TIER2_GEN__BUDGET__RMB_HARD_LIMIT", + "TIER2_GEN__GATES__QPASS_GO_MIN", + "TIER2_GENCONFIG", + "TIER2_GENCONFIG_NACOS", + "TIER2_GENCONFIG_NACOS_DATAID", + "TIER2_GENCONFIG_NACOS_GROUP", + "NACOS_SERVER_ADDR", + "NACOS_NAMESPACE", + "NACOS_USERNAME", + "NACOS_PASSWORD", + "NO_PROXY", + "no_proxy", +) + + +@pytest.fixture() +def clean_genconfig(): + """每测前后:detach 热源 + reload YAML 缓存 + 清相关 env,保证从纯本地基线起跑、测后不泄漏。""" + genconfig.detach_hot_source() + genconfig.reload() + snap = {k: os.environ.get(k) for k in _ENV_KEYS} + for k in _ENV_KEYS: + os.environ.pop(k, None) + try: + yield + finally: + genconfig.detach_hot_source() + genconfig.reload() + for k, v in snap.items(): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v + + +class _FakeHotSource: + """最小热源:满足 genconfig 期望的 get(key, default)->value 契约(NacosHotConfig 读接口的子集)。 + + genconfig 用 src.get("area.key", _MISSING) 查:store 无该键时把 genconfig 传入的 sentinel 原样返回, + genconfig 据此判「热源没这个键」继续回落。故这里 default 必须原样透传、不能替换成别的默认。 + """ + + def __init__(self, store: dict, raise_on_get: bool = False): + self._store = dict(store) + self._raise = raise_on_get # True 模拟热源取值异常(get() 抛) + + def get(self, key, default=None): + if self._raise: + raise RuntimeError("hot source boom") + return self._store.get(key, default) + + +class _FakeNacosClient: + """fake v1 nacos client:喂给真 NacosHotConfig,验路径②③(不可达 / 读不到)。契约对齐已核验的 v1。""" + + def __init__(self, content, unreachable: bool = False): + self._content = content + self.unreachable = unreachable # True → get_config 抛(模拟连磁盘快照都没有的不可达) + + def get_config(self, data_id, group, timeout=None, no_snapshot=None): + if self.unreachable: + raise RuntimeError("nacos unreachable") + return self._content # None 模拟「订阅的空间/dataId 收不到值」(namespace 配错表现) + + def add_config_watcher(self, data_id, group, cb, content=None): + pass # 订阅 no-op:这些回落测试不涉及推送热更 + + +# ════════════════════════════════════════════════════════════════════════════ +# 回落链四条(设计 §4 步骤4 验证要求) +# ════════════════════════════════════════════════════════════════════════════ + +def test_path1_not_attached_uses_local_chain(clean_genconfig): + """① 未 attach → genconfig 完全走原链(env→YAML→内置默认),行为 = 现状字节不变。""" + assert genconfig._hot_source is None # 默认未接热源 + baseline = genconfig.get("budget", "rmb_hard_limit", 50.0) # 纯本地基线(真 YAML 或内置默认) + # 设 env 一次性覆盖仍照常生效(未 attach 时 env 就是最高优先级)。 + os.environ["TIER2_GEN__BUDGET__RMB_HARD_LIMIT"] = "88.0" + assert genconfig.get("budget", "rmb_hard_limit", 50.0) == 88.0 + assert baseline != 88.0 # 基线不是 env 值,证明确实走了本地链 + + +def test_path1b_attached_but_key_absent_falls_back_to_env(clean_genconfig): + """① 热源已 attach 但缺该键 → 回落 env(热源无值不把取值截断成 None)。""" + genconfig.attach_hot_source(_FakeHotSource({"gates.qpass_go_min": 0.9})) # 只有别的键 + os.environ["TIER2_GEN__BUDGET__RMB_HARD_LIMIT"] = "77.0" + # 热源没有 budget.rmb_hard_limit → 回落 env 77.0(不是热源那条 gates 键、更不是 None)。 + assert genconfig.get("budget", "rmb_hard_limit", 50.0) == 77.0 + + +def test_path2_nacos_unreachable_falls_back_to_local(clean_genconfig): + """② Nacos 不可达:真 NacosHotConfig + 不可达 client → store 空 → genconfig 回落本地。""" + genconfig.detach_hot_source() + baseline = genconfig.get("budget", "rmb_hard_limit", 50.0) # 先取纯本地基线 + fake = _FakeNacosClient(None, unreachable=True) + hot = NacosHotConfig(client=fake, data_id="gen-hot-params-tier2", group="DEFAULT_GROUP", defaults={}) + hot.start() # 不可达:get_config 抛,best-effort 留空 store,不抛 + genconfig.attach_hot_source(hot) + # 热源 store 空(不可达兜底)→ genconfig 回落本地基线,不被 Nacos 连累。 + assert genconfig.get("budget", "rmb_hard_limit", 50.0) == baseline + + +def test_path3_namespace_misconfig_reads_nothing_falls_back(clean_genconfig): + """③ namespace 配错:订阅空间/dataId 收不到值(get_config 返 None)→ store 空 → 回落本地。 + + 佐证前置任务:写侧 publish 进真实 namespaceId、读侧订阅另一个(如误写 public)→ 永远读不到, + 表现就是本条——store 停在空、genconfig 回落本地。故写读两侧 namespaceId 必须一次性对齐。 + """ + genconfig.detach_hot_source() + baseline = genconfig.get("budget", "rmb_hard_limit", 50.0) + fake = _FakeNacosClient(None) # 非不可达,但订阅的空间读不到任何内容 + hot = NacosHotConfig(client=fake, data_id="gen-hot-params-tier2", group="DEFAULT_GROUP", defaults={}) + hot.start() + genconfig.attach_hot_source(hot) + assert genconfig.get("budget", "rmb_hard_limit", 50.0) == baseline + + +def test_path4_yaml_fallback_when_hot_and_env_absent(clean_genconfig): + """④ YAML 回落:热源无该键、无 env → 取本地 YAML 与内置默认(结果 = 纯本地基线)。""" + genconfig.detach_hot_source() + baseline = genconfig.get("budget", "rmb_hard_limit", 50.0) + genconfig.attach_hot_source(_FakeHotSource({})) # 空热源(无任何键) + assert genconfig.get("budget", "rmb_hard_limit", 50.0) == baseline # 回落本地,与未 attach 一致 + + +# ════════════════════════════════════════════════════════════════════════════ +# 优先级 / 两档隔离 / 门阈值 / 脏值 +# ════════════════════════════════════════════════════════════════════════════ + +def test_hot_source_overrides_env_and_yaml(clean_genconfig): + """优先级 Nacos > env > YAML:热源命中即压过 env(配置中心才是权威,遗留 env 不得静默盖住激活值)。""" + os.environ["TIER2_GEN__BUDGET__RMB_HARD_LIMIT"] = "77.0" # 机器上遗留实验 env + genconfig.attach_hot_source(_FakeHotSource({"budget.rmb_hard_limit": 99.0})) + assert genconfig.get("budget", "rmb_hard_limit", 50.0) == 99.0 # 热源赢 + + +def test_two_tiers_read_own_dataid(clean_genconfig): + """两档各读各 dataId:tier2 进程热源只有 tier2 键、cheap 进程热源只有 cheap 键,互不串档。""" + # —— tier2 进程:先取 cheap 键的纯本地基线(fixture 保证起始未 attach),再 attach tier2 dataId 的热源 —— + cheap_local = genconfig.get("budget", "cheap_rmb_hard_limit", 10.0) # cheap 键纯本地基线 + genconfig.attach_hot_source(_FakeHotSource({"budget.rmb_hard_limit": 55.0})) + assert genconfig.get("budget", "rmb_hard_limit", 50.0) == 55.0 # tier2 键命中热源 + assert genconfig.get("budget", "cheap_rmb_hard_limit", 10.0) == cheap_local # cheap 键热源没有→回落本地 + + # —— cheap 进程:换成 attach cheap dataId 的热源(store 里是 budget.cheap_rmb_hard_limit)—— + genconfig.detach_hot_source() + tier2_baseline = genconfig.get("budget", "rmb_hard_limit", 50.0) # tier2 键的纯本地基线 + genconfig.attach_hot_source(_FakeHotSource({"budget.cheap_rmb_hard_limit": 12.0})) + assert genconfig.get("budget", "cheap_rmb_hard_limit", 10.0) == 12.0 # cheap 键命中热源 + assert genconfig.get("budget", "rmb_hard_limit", 50.0) == tier2_baseline # tier2 键热源没有→回落本地 + + +def test_gate_thresholds_also_hot_read(clean_genconfig): + """门阈值(gates 区)与软预算(budget 区)一样吃上热读——路 B 覆盖预算 + 阈值两类。""" + genconfig.attach_hot_source(_FakeHotSource({ + "budget.rmb_hard_limit": 66.0, + "gates.qpass_go_min": 0.66, + })) + assert genconfig.get("budget", "rmb_hard_limit", 50.0) == 66.0 + assert genconfig.get("gates", "qpass_go_min", 0.40) == 0.66 + + +def test_hot_value_coerced_by_type(clean_genconfig): + """热源值按内置默认类型转型:JSON 里的字符串 "45" 落到 float 旋钮上得 45.0(与 env/YAML 同款转型)。""" + genconfig.attach_hot_source(_FakeHotSource({"budget.rmb_hard_limit": "45"})) + got = genconfig.get("budget", "rmb_hard_limit", 50.0) + assert got == 45.0 and isinstance(got, float) + + +def test_dirty_hot_value_falls_back_no_throw(clean_genconfig): + """脏热值(转不动的字符串)→ 保旧回落本地、不抛、不穿透成 None。""" + genconfig.detach_hot_source() + baseline = genconfig.get("budget", "rmb_hard_limit", 50.0) + genconfig.attach_hot_source(_FakeHotSource({"budget.rmb_hard_limit": "not-a-number"})) + assert genconfig.get("budget", "rmb_hard_limit", 50.0) == baseline # 脏值被挡,回落本地 + + +def test_hot_source_get_exception_falls_back(clean_genconfig): + """热源 get 自身抛异常 → genconfig 视为「没取到」回落本地,绝不连累主链。""" + genconfig.detach_hot_source() + baseline = genconfig.get("budget", "rmb_hard_limit", 50.0) + genconfig.attach_hot_source(_FakeHotSource({"budget.rmb_hard_limit": 99.0}, raise_on_get=True)) + assert genconfig.get("budget", "rmb_hard_limit", 50.0) == baseline + + +# ════════════════════════════════════════════════════════════════════════════ +# 工厂(build_and_attach_hot_source) +# ════════════════════════════════════════════════════════════════════════════ + +def test_factory_disabled_by_default(clean_genconfig): + """工厂旗默认关:不设 TIER2_GENCONFIG_NACOS → build 返回 None、不 attach、genconfig 走原链。""" + assert genconfig_nacos.is_enabled() is False + hot = genconfig_nacos.build_and_attach_hot_source(tier="tier2") + assert hot is None + assert genconfig._hot_source is None + + +def test_factory_attaches_and_genconfig_reads_hot(clean_genconfig): + """启用旗打开 + fake client 有值 → 工厂建 client → NacosHotConfig 初读 → attach → genconfig 命中热值。""" + os.environ["TIER2_GENCONFIG_NACOS"] = "1" + fake_client = _FakeNacosClient(json.dumps({"budget.rmb_hard_limit": 42.0})) + with mock.patch("nacos.NacosClient", return_value=fake_client): + hot = genconfig_nacos.build_and_attach_hot_source(tier="tier2") + assert hot is not None + assert genconfig._hot_source is hot + assert genconfig.get("budget", "rmb_hard_limit", 50.0) == 42.0 # 端到端:经工厂 attach 后读到热值 + + +def test_factory_best_effort_on_client_error(clean_genconfig): + """工厂 best-effort:client 构造抛 → build 返回 None、不抛、genconfig 未 attach、回落纯本地。""" + os.environ["TIER2_GENCONFIG_NACOS"] = "1" + with mock.patch("nacos.NacosClient", side_effect=RuntimeError("nacos unreachable")): + hot = genconfig_nacos.build_and_attach_hot_source(tier="tier2") # 不应抛 + assert hot is None + assert genconfig._hot_source is None + + +def test_factory_dataid_per_tier(clean_genconfig): + """按档隔离:tier="cheap" 订阅 gen-hot-params-cheap、tier="tier2" 订阅 gen-hot-params-tier2。""" + os.environ["TIER2_GENCONFIG_NACOS"] = "1" + seen = {} + + def _spy_hotconfig(*, client, data_id, group, defaults): + seen["data_id"] = data_id + seen["defaults"] = defaults + return mock.MagicMock() + + # cheap 档 + with mock.patch("nacos.NacosClient", return_value=mock.MagicMock()), \ + mock.patch("worker.nacos_hotconfig.NacosHotConfig", side_effect=_spy_hotconfig): + genconfig_nacos.build_and_attach_hot_source(tier="cheap") + assert seen["data_id"] == "gen-hot-params-cheap" + assert seen["defaults"] == {} # 坑④:热源 defaults 必须空,不引入第二默认值 + + genconfig.detach_hot_source() + # tier2 档 + with mock.patch("nacos.NacosClient", return_value=mock.MagicMock()), \ + mock.patch("worker.nacos_hotconfig.NacosHotConfig", side_effect=_spy_hotconfig): + genconfig_nacos.build_and_attach_hot_source(tier="tier2") + assert seen["data_id"] == "gen-hot-params-tier2" + + +def test_factory_dataid_env_override(clean_genconfig): + """env TIER2_GENCONFIG_NACOS_DATAID 显式覆盖档默认(演示脚本用它指向演示专用 dataId)。""" + os.environ["TIER2_GENCONFIG_NACOS"] = "1" + os.environ["TIER2_GENCONFIG_NACOS_DATAID"] = "gen-hot-params-demo" + seen = {} + + def _spy_hotconfig(*, client, data_id, group, defaults): + seen["data_id"] = data_id + return mock.MagicMock() + + with mock.patch("nacos.NacosClient", return_value=mock.MagicMock()), \ + mock.patch("worker.nacos_hotconfig.NacosHotConfig", side_effect=_spy_hotconfig): + genconfig_nacos.build_and_attach_hot_source(tier="tier2") # 档默认本应 tier2,被 env 覆盖 + assert seen["data_id"] == "gen-hot-params-demo" + + +def test_factory_proxy_bypass_before_client_build(clean_genconfig): + """内网铁律:NO_PROXY 旁路必须在【建 client 前】完成 —— 构造 NacosClient 时 NO_PROXY 已含 Nacos host。""" + os.environ["TIER2_GENCONFIG_NACOS"] = "1" + os.environ["NACOS_SERVER_ADDR"] = "100.64.0.8:8848" + seen = {} + + def _spy_ctor(*_a, **_kw): + seen["no_proxy_at_build"] = os.environ.get("NO_PROXY", "") + return mock.MagicMock() + + with mock.patch("nacos.NacosClient", side_effect=_spy_ctor), \ + mock.patch("worker.nacos_hotconfig.NacosHotConfig", return_value=mock.MagicMock()): + genconfig_nacos.build_and_attach_hot_source(tier="tier2") + assert "100.64.0.8" in seen.get("no_proxy_at_build", "") + + +def test_factory_namespace_defaults_empty_string(clean_genconfig): + """真实 namespaceId 默认空串(勿写 public;前置任务口径)——构造 client 时 namespace 关键字 == ""。""" + os.environ["TIER2_GENCONFIG_NACOS"] = "1" + ctor = mock.MagicMock(return_value=mock.MagicMock()) + with mock.patch("nacos.NacosClient", ctor), \ + mock.patch("worker.nacos_hotconfig.NacosHotConfig", return_value=mock.MagicMock()): + genconfig_nacos.build_and_attach_hot_source(tier="tier2") + _args, kwargs = ctor.call_args + assert kwargs.get("namespace") == "" # 默认公共空间 = 空串 + assert kwargs.get("username") == "nacos" + + +def test_rollback_switch_detach(clean_genconfig): + """回退开关:detach 后 genconfig 回落纯本地链(设计 §3.6/§6 一处开关切回本地 YAML)。""" + genconfig.attach_hot_source(_FakeHotSource({"budget.rmb_hard_limit": 99.0})) + assert genconfig.get("budget", "rmb_hard_limit", 50.0) == 99.0 + genconfig_nacos.detach() # 回退 + assert genconfig._hot_source is None + assert genconfig.get("budget", "rmb_hard_limit", 50.0) != 99.0 # 回落本地,不再是热值 diff --git a/tier2/gen-worker/worker/genconfig.py b/tier2/gen-worker/worker/genconfig.py index d96b2eb6..192138e1 100644 --- a/tier2/gen-worker/worker/genconfig.py +++ b/tier2/gen-worker/worker/genconfig.py @@ -7,8 +7,14 @@ 调生成 = 改 YAML 重跑,不改码、不重同步。这正是图说 B 族「配置注册表(唯一事实源 / 版本化 / 运行时读)」 在生成器旋钮这一面的最小落地——配置是真相,代码只是按 id 取值的消费方。 -【取值三级回落(运行时读,失败绝不中断主链)】 +【取值优先级(运行时读,失败绝不中断主链)】 get(area, key, default) 每次调用按以下顺序取值,任一级取到即用: + ⓪ Nacos 热源(配置控制面阶段二接入,默认不启用):worker 启动期若由工厂 genconfig_nacos + .build_and_attach_hot_source() attach 了一个 NacosHotConfig(env TIER2_GENCONFIG_NACOS 启用), + 则最先查它——配置中心激活的预算/门阈值经 Nacos 长轮询推送热读进来,改一次配置中心、下一次生成 + 即读到新值,不重启。未启用(默认)时本级不存在,取值链 = 原三级,行为与阶段二前逐字节一致。 + 热源以复合键 "area.key" 存值(genconfig 是 (area,key) 二维,复合键无歧义扁平化);缺该键则继续 + 往下回落,绝不引入第二默认值(热源只给「有就用」的值,缺失全交下面 env/YAML/内置默认)。 ① env 单点覆盖:环境变量 TIER2_GEN____(全大写、双下划线分隔)——临时压一个值不必改文件, 适合「这一批我想试 max_iters=60」这种一次性实验(改完 unset 即恢复)。 ② 外部 YAML:tier2/config/generation.yaml(路径可被 env TIER2_GENCONFIG 指向别处)的 area.key。 @@ -154,6 +160,41 @@ _cache_path: str | None = None _warned: set[str] = set() +# ── Nacos 热源(配置控制面阶段二步骤4:genconfig 读取后端提升到 Nacos)──────────────────── +# 工厂 genconfig_nacos.build_and_attach_hot_source() 在 worker 启动期把一个 NacosHotConfig 实例 +# attach 进这个进程级单例槽;get() 取值链最前端先查它(见 get() 的 ⓪ 层)。未 attach(默认)→ get() +# 完全走原链(env→YAML→内置默认→调用方 default),行为与阶段二前逐字节一致。 +# 为什么是进程级单例、且 genconfig 不认档:cheap Service 与 tier2 Service 是两个独立进程,却都 +# `from worker import genconfig` 共用本模块。两档隔离不靠 genconfig 区分,而靠【各档进程 attach 各档 +# dataId 的热源】——cheap 进程 attach cheap dataId 的 NacosHotConfig(store 里是 budget.cheap_* 键)、 +# tier2 进程 attach tier2 dataId 的(store 里是 budget.rmb_hard_limit 等)。genconfig 只查「当前进程 +# attach 的那个热源」,故同一份 get() 代码在两档进程里各取各的 dataId,天然不串档。 +_hot_source: Any = None + +# sentinel:区分「热源里没有这个键」与「热源里键值恰好是 None」。热源缺键必须继续回落到 env/YAML/ +# 内置默认,不能被当成一个 None 值截断(否则会吞掉下面所有回落级)。 +_MISSING: Any = object() + + +def attach_hot_source(source: Any) -> None: + """worker 启动期由工厂注入 Nacos 热源(NacosHotConfig 实例);此后 get() 最先查它。 + + 幂等可重入:重复 attach 覆盖为最新源。source 只需满足 get(key, default)->value 契约 + (NacosHotConfig 即是),genconfig 不依赖其内部实现——故单测可注入任意 fake 源。 + """ + global _hot_source + _hot_source = source + + +def detach_hot_source() -> None: + """卸载热源,get() 回落纯本地链(env→YAML→内置默认→调用方 default)。 + + 这是设计 §3.6/§6「回退开关」在进程内的一处落点:热源出问题时切回纯本地,调用面一行不改。 + """ + global _hot_source + _hot_source = None + + def _warn_once(tag: str, msg: str) -> None: """同一 tag 只告警一次(防 best-effort 路径在循环里刷屏);可追溯日志,带 [tier2-genconfig] 前缀。""" if tag in _warned: @@ -284,7 +325,31 @@ def get(area: str, key: str, default: Any = None) -> Any: # 转型基准:内置默认有该 key 用它的类型,否则用 default 的类型(都为 None 则不强转)。 like = builtin.get(key, default) - # ① env 单点覆盖(最高优先级,便于一次性实验,不必改文件)。 + # ⓪ Nacos 热源(配置控制面阶段二:最高优先级)。未 attach(默认)/ 热源无此键 / 取值异常 → 静默 + # 继续回落,绝不引入第二默认值(热源只给「有就用」的值,缺失一律交下面 env/YAML/内置默认)。 + # 复合键 area.key:把 genconfig 的 (area,key) 二维无歧义扁平化成热源 store 的一个键;激活编排 / + # 演示脚本按同一 "area.key" 组 dataId JSON。热源命中即返回,根本不读 YAML —— mtime 缓存无从盖住热值。 + src = _hot_source + if src is not None: + try: + hot_val = src.get(f"{area}.{key}", _MISSING) + except Exception as exc: # noqa: BLE001 —— 热源取值绝不连累主链:异常即视为「没取到」回落 + hot_val = _MISSING + _warn_once( + f"hot-err:{area}.{key}", + f"Nacos 热源取 {area}.{key} 异常({type(exc).__name__}: {exc})→ 回落 env/YAML/默认。", + ) + if hot_val is not _MISSING: + coerced = _coerce(hot_val, like) + if coerced is not None: + return coerced + # 热源有该键但值转不动(类型脏)→ 告警回落(与 env/YAML 脏值同处置,不让脏热值穿透)。 + _warn_once( + f"hot-bad:{area}.{key}", + f"Nacos 热源 {area}.{key}={hot_val!r} 无法转成 {type(like).__name__} → 忽略,回落 env/YAML/默认。", + ) + + # ① env 单点覆盖(Nacos 热源之下、YAML 之上;一次性实验逃生舱,不必改文件)。 env_name = f"{_ENV_PREFIX}{area.upper()}__{key.upper()}" env_raw = os.environ.get(env_name) if env_raw is not None: diff --git a/tier2/gen-worker/worker/genconfig_nacos.py b/tier2/gen-worker/worker/genconfig_nacos.py new file mode 100644 index 00000000..dedaffcc --- /dev/null +++ b/tier2/gen-worker/worker/genconfig_nacos.py @@ -0,0 +1,156 @@ +"""genconfig_nacos.py —— 把 genconfig 的读取后端提升到 Nacos 的启动期工厂(配置控制面阶段二·步骤4)。 + +【这份解决什么问题】 +阶段〇建好了 nacos_hotconfig.NacosHotConfig(只读视图:订阅一个 Nacos dataId、长轮询热更、线程安全读), +但它「只消费已建好的 v1 NacosClient、自己不建 client」。阶段二步骤4 把这最后一段接起来:本模块是 +**worker 启动期**的单例工厂,负责把 v1 client 构造齐(server / 鉴权 / NO_PROXY 旁路 / 真实 namespaceId / +group / 按档 dataId),注入 NacosHotConfig,start() 订阅,再 attach 进 genconfig 作「热源」。此后进程内 +所有 genconfig.get(area,key) 调用最先查 Nacos 热源——middleware 的软预算目标、门判阈值都自动吃上配置 +中心的激活值,调用面一行不改。 + +【显式启用开关(默认关 = 现状字节不变)】 +env TIER2_GENCONFIG_NACOS ∈ {1,true,yes,on} 才启用;不设(默认)时本工厂 no-op、不 attach、不牵 nacos +依赖,genconfig 完全走原链(env→YAML→内置默认),行为与阶段二前逐字节一致。这个旗同时是设计 §3.6/§6 +的【回退开关】:线上出问题就关旗重启,genconfig 立刻切回纯本地 YAML(运行时也可 genconfig.detach_hot_source())。 + +【按档隔离的 dataId(设计 §3.3)】 +cheap-worker 与 tier2 gen-worker 是两个独立进程、各起一个本工厂:tier="tier2" 订阅 gen-hot-params-tier2、 +tier="cheap" 订阅 gen-hot-params-cheap。两档各 attach 各自 dataId 的热源,publish 一档不冲掉另一档的键。 +genconfig 本身不认档(见 genconfig._hot_source 注释),隔离全靠「哪个进程订阅哪个 dataId」。 + +【NO_PROXY 内网铁律(阶段〇同款坑)】 +本机系统代理是 fake-ip 198.18.x;nacos client 走 HTTP 直连内网 100.64.0.8:8848 必须绕过它,且必须在 +【建 client 前】把 Nacos host 塞进 NO_PROXY/no_proxy(urllib 建连时读环境,晚设不生效,否则栽 502)。与 +service/nacos_registry._bypass_proxy_for 同款;此处 worker 包内自带一份(worker 是底层,不反向依赖 service 包)。 + +【真实 namespaceId = 空串(前置任务口径)】 +默认公共空间的 namespaceId 是空串,不是字面量 "public"(显示名≠id)。写侧(yudao publish)与读侧(本工厂) +必须对齐同一真实 namespaceId,否则激活写进一个空间、worker 在另一个空间读、配置永远推不到(前置任务 056433d5 +已把 nacos_registry / Java / 凭据档统一到空串,本工厂沿同口径:NACOS_NAMESPACE 默认空串)。 + +【绝不引入第二默认值(坑④·tech-decisions §13)】 +注入 NacosHotConfig 的 defaults 刻意为空 {}:热源缺某键时 NacosHotConfig.get 返回 genconfig 传入的 +sentinel,由 genconfig 回落链(env→YAML→内置默认)接管——热源侧登记默认值 = 第二默认源,会静默盖住单源 +(双源默认值影蔽刚在收敛环栽过),故这里 defaults 必须空。 + +【best-effort 铁律】 +构造 / 订阅 全程 try/except:启用了但 Nacos 不可达 / 鉴权失败 / 缺 nacos 库 → 只告警、不抛、不 attach, +genconfig 回落纯本地链,生成主链不被配置中心连累。 +""" + +from __future__ import annotations + +import os +from typing import TYPE_CHECKING, Any + +# genconfig 顶层零重依赖(不牵 agentscope),安全直接 import。包内相对导入优先,PYTHONPATH=gen-worker 直跑兜底。 +try: + from . import genconfig # 包内相对导入(worker.genconfig_nacos) +except ImportError: # pragma: no cover —— PYTHONPATH=gen-worker 直跑时的绝对导入兜底 + from worker import genconfig # type: ignore + +if TYPE_CHECKING: + from worker.nacos_hotconfig import NacosHotConfig + + +# 启用旗(默认关 = 现状不变;同时是设计 §3.6/§6 回退开关)。与 genconfig 的 TIER2_GEN__ 前缀区分: +# TIER2_GEN__ 是「压某个旋钮值」,本旗是「是否把读取后端接到 Nacos」的总开关。 +_ENABLE_FLAG = "TIER2_GENCONFIG_NACOS" + +# 按档默认 dataId(设计 §3.3 按档隔离)。env TIER2_GENCONFIG_NACOS_DATAID 显式指定则不分档、直接用它 +# (演示脚本用它指向演示专用 dataId,不碰生产 dataId)。 +_DEFAULT_DATAIDS = { + "tier2": "gen-hot-params-tier2", + "cheap": "gen-hot-params-cheap", +} + +_TRUE = ("1", "true", "yes", "on") + + +def is_enabled() -> bool: + """启用旗是否打开(默认关)。调用方可先查再决定是否 build;也是回退开关的判据。""" + return os.environ.get(_ENABLE_FLAG, "").strip().lower() in _TRUE + + +def _bypass_proxy_for(host: str) -> None: + """把 host 并进 NO_PROXY/no_proxy(内网直连绕系统 fake-ip 代理;必须在建 nacos client 前调)。 + + 与 service/nacos_registry._bypass_proxy_for 同款逻辑(worker 是底层,不反向 import service 包,故各存 + 一份):两个大小写变体都写;逗号分隔精确比对各段,避免 "100.64.0.8" 与 "100.64.0.80" 之类子串误判; + 已在列表就不重复追加。 + """ + for key in ("NO_PROXY", "no_proxy"): + cur = os.environ.get(key, "") + 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 _resolve_data_id(tier: str) -> str: + """按档解析 dataId:env TIER2_GENCONFIG_NACOS_DATAID 显式覆盖 > 档默认 > 兜底 tier2 档名(未知档不炸)。""" + override = os.environ.get("TIER2_GENCONFIG_NACOS_DATAID", "").strip() + if override: + return override + return _DEFAULT_DATAIDS.get(tier, _DEFAULT_DATAIDS["tier2"]) + + +def build_and_attach_hot_source(*, tier: str = "tier2") -> "NacosHotConfig | None": + """worker 启动期调:构造 v1 NacosClient → 注入 NacosHotConfig → start() → attach 进 genconfig。 + + Args: + tier: 档位("tier2" / "cheap"),决定订阅哪个 dataId(设计 §3.3 按档隔离)。 + + Returns: + 已 attach 的 NacosHotConfig(供调用方持有 / 关停 / 测试);启用旗未开或构造失败 → 返回 None + (genconfig 走原链)。全程 best-effort:任何异常只告警、不抛、不 attach。 + + 注入 NacosHotConfig 的 defaults 刻意为空 {}:热源缺某键时 get 返回 sentinel,由 genconfig 回落链 + (env→YAML→内置默认)接管——热源侧【绝不】登记第二套默认值(坑④双源默认值影蔽,tech-decisions §13)。 + """ + if not is_enabled(): + # 默认关:不 attach、不牵 nacos 依赖,genconfig 走原链(现状字节不变)。 + return None + + nacos_addr = os.environ.get("NACOS_SERVER_ADDR", "100.64.0.8:8848") + namespace = os.environ.get("NACOS_NAMESPACE", "") # 默认公共空间 = 空串(勿写 public;前置任务口径) + username = os.environ.get("NACOS_USERNAME", "nacos") + password = os.environ.get("NACOS_PASSWORD", "") + group = os.environ.get("TIER2_GENCONFIG_NACOS_GROUP", "DEFAULT_GROUP") + data_id = _resolve_data_id(tier) + + # 内网铁律:必须在【建 client 前】把 Nacos host 塞进 NO_PROXY,否则 HTTP 直连被 fake-ip 代理拦(502)。 + _bypass_proxy_for(nacos_addr.split(":")[0]) + + try: + import nacos # noqa: PLC0415 —— 惰性 import:默认关时不牵 nacos 依赖(本地/CLI/6c6g 免装) + from worker.nacos_hotconfig import NacosHotConfig # noqa: PLC0415 + + client = nacos.NacosClient( + nacos_addr, + namespace=namespace, + username=username, + password=password, + ) + # defaults={} 是硬约束:热源不登记默认值,缺键交 genconfig 回落链(杜绝双源默认值影蔽,坑④)。 + hot = NacosHotConfig(client=client, data_id=data_id, group=group, defaults={}) + hot.start() # 初读 + 订阅长轮询推送(best-effort:Nacos 不可达也不抛,留空 store 让 genconfig 回落) + genconfig.attach_hot_source(hot) + print( + f"[genconfig-nacos] 已接 Nacos 热源:tier={tier} dataId={data_id} group={group} " + f"namespace={namespace!r} nacos={nacos_addr}", + flush=True, + ) + return hot + except Exception as e: # noqa: BLE001 —— best-effort:接线失败不 attach、不抛,genconfig 回落纯本地链 + print( + f"[genconfig-nacos] 接 Nacos 热源失败(genconfig 回落纯本地 YAML,best-effort):" + f"{type(e).__name__}: {e}", + flush=True, + ) + return None + + +def detach() -> None: + """卸载热源,genconfig 回落纯本地链(设计 §3.6/§6 回退开关的显式入口;转调 genconfig.detach_hot_source)。""" + genconfig.detach_hot_source()