games-development-ai/tier2/gen-worker/scripts/demo_nacos_hot_read.py

#!/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)