执行摘要

• 一句话：FPM 端点写入迁移至配置 bags，消除 ServerArgs 消息总线
• 推荐动作：值得快速阅读，特别关注 _init_fpm() 中把 ServerArgs 从消息总线降级为纯配置持有者的写法，以及测试如何用真实 override_server_args 发布配置来覆盖生产路径。这是理解 sglang 配置治理（bags、runtime_context、writer ratchet）演进的好范例，但本身改动很小，不需要精读。

功能与动机

PR body 明确指出：当 --forward-pass-metrics-ipc-name 未设置时，reporter 生成的端点需要交给外部消费者，文档约定（server_arguments.mdx）是消费者从 server config 读回 server_args.forward_pass_metrics_ipc_name，而该读回路径是 scheduler 的 get_internal_state，它已经报告 get_context().resolved_server_args_dict()（发布后覆盖叠加的精净配置）。因此写入应移动到 get_context().override(...)，读取应紧邻写入放到 get_observability()。同时说明这组变更属于五部分系列（Part 4 of five），且替代了因 GitHub 链式 base 栈问题而关闭的 #33241。

实现拆解

1. 写入路径迁移（metrics_reporter.py）：在 _init_fpm() 中，将 base_endpoint = self.scheduler.server_args.forward_pass_metrics_ipc_name 改为 base_endpoint = get_observability().forward_pass_metrics_ipc_name；将 self.scheduler.server_args.override("metrics_reporter.ipc_endpoint", forward_pass_metrics_ipc_name=base_endpoint) 改为 get_context().override(...)，并新增 get_context, get_observability 导入。这样端点写入进入 resolved-config bags，实例字段不再承载 post-publish 变更。

2. 读取路径对齐（metrics_reporter.py）：_init_fpm() 随后仍用 base_endpoint 拼接 .{dp_rank} 后缀构造 PUB 端点并启动 _FpmPublisherThread，外部消费者通过 get_context().resolved_server_args_dict() 读回同一值，保证写入/读取在同一条 bag 通道上。

3. 测试夹具重构（test_forward_pass_metrics.py）：将手搓的 _fake_server_args（SimpleNamespace + 自定义 _override 函数）替换为 _publish_server_args(test, **fields)，内部使用 get_context().override_server_args(**fields) + override.install() 发布真实配置，并通过 test.addCleanup(override.restore) 做清理；_make_reporter 和所有调用点增加 test 参数。核心断言更新为：get_observability().forward_pass_metrics_ipc_name 以 ipc:// 开头，且与 get_context().resolved_server_args_dict()["forward_pass_metrics_ipc_name"] 相等，同时断言 scheduler.server_args.forward_pass_metrics_ipc_name 为 None，验证实例不再被写入。

4. writer ratchet 基线下调（test_server_args_writer_ratchet.py）：_BASELINE 从 19 降到 18，锁定本 PR 移除一个 server_args.override( 调用点的进展。

5. 配套验证：PR body 说明 test/registered/unit/observability 与 config ratchet 测试通过，完整注册 CPU 套件相对 base commit 无新增失败。

关键文件：

• python/sglang/srt/managers/scheduler_components/metrics_reporter.py（模块 指标上报；类别 source；类型 dependency-wiring；符号 _init_fpm）: 这是 FPM 端点的写入口，_init_fpm() 从 server_args.override() 改为 get_context().override()，读路径同步改为 get_observability()，是该 PR 的核心源码变更，决定了端点进入 resolved-config bags。
• test/registered/unit/observability/test_forward_pass_metrics.py（模块 指标上报；类别 test；类型 test-coverage；符号 _publish_server_args, _make_reporter）: 测试从 SimpleNamespace 手搓替身改为通过 override_server_args 发布真实配置，并新增对 get_observability() 与 resolved_server_args_dict() 的断言，直接验证 bags 写入契约。
• test/registered/unit/test_server_args_writer_ratchet.py（模块 配置治理；类别 test；类型 test-coverage）: writer ratchet 基线从 19 降到 18，锁定本 PR 移除一个 server_args.override() 调用点的进展，防止未来回退。

关键符号：_init_fpm, _publish_server_args, _make_reporter

关键源码片段

python/sglang/srt/managers/scheduler_components/metrics_reporter.py

这是 FPM 端点的写入口，_init_fpm() 从 server_args.override() 改为 get_context().override()，读路径同步改为 get_observability()，是该 PR 的核心源码变更，决定了端点进入 resolved-config bags。

# python/sglang/srt/managers/scheduler_components/metrics_reporter.py
# _init_fpm 是 FPM publisher 的初始化入口。
# 关键变化：端点生成后不再写入 scheduler.server_args 实例，
# 而是通过 get_context().override() 写入 resolved-config bags，
# 让外部消费者能从 get_context().resolved_server_args_dict() 读回。
def _init_fpm(self):
    """Initialize Forward Pass Metrics (FPM) publisher if configured."""
    self.scheduler.enable_fpm = False
    if (
        self.scheduler.server_args.enable_forward_pass_metrics
        and self.scheduler.ps.attn_tp_rank == 0
        and self.scheduler.ps.pp_rank == self.scheduler.ps.pp_size - 1
    ):
        from sglang.srt.observability.forward_pass_metrics import _FpmPublisherThread
​
        self.scheduler._fpm_dp_rank = (
            self.scheduler.ps.dp_rank if self.scheduler.ps.dp_rank is not None else 0
        )
        self.scheduler._fpm_worker_id = self.scheduler.server_args.forward_pass_metrics_worker_id
​
        # 读取改为走 get_observability()，与写入处于同一条 bag 通道
        base_endpoint = get_observability().forward_pass_metrics_ipc_name
        if base_endpoint is None:
            # 未显式指定时生成临时 IPC 路径作为端点
            ipc_path = tempfile.NamedTemporaryFile(delete=False).name
            base_endpoint = f"ipc://{ipc_path}"
            # 写入改为 get_context().override()，而不是 server_args.override()
            get_context().override(
                "metrics_reporter.ipc_endpoint",
                forward_pass_metrics_ipc_name=base_endpoint,
            )
​
        # 每个 DP rank 一个独立 endpoint，后续拼接 rank 后缀
        endpoint = f"{base_endpoint}.{self.scheduler._fpm_dp_rank}"
        self.scheduler._fpm_publisher = _FpmPublisherThread(
            endpoint,
            worker_id=self.scheduler._fpm_worker_id,
            dp_rank=self.scheduler._fpm_dp_rank,
        )
        self.scheduler._fpm_gpu_time_acc = 0.0
​
        def _fpm_device_timer_reporter(t, **_kwargs):
            self.scheduler._fpm_gpu_time_acc += t
​
        if self.forward_pass_device_timer is not None:
            self.forward_pass_device_timer.add_reporter(_fpm_device_timer_reporter)
        else:
            self.forward_pass_device_timer = DeviceTimer(reporter=_fpm_device_timer_reporter)
​
        self.scheduler._fpm_uses_device_timer = True
        self.scheduler.enable_fpm = True
        logger.info(
            "FPM: ZMQ PUB bound on %s (dp_rank=%d, device_timer=%s)",
            endpoint,
            self.scheduler._fpm_dp_rank,
            self.scheduler._fpm_uses_device_timer,
        )

test/registered/unit/observability/test_forward_pass_metrics.py

测试从 SimpleNamespace 手搓替身改为通过 override_server_args 发布真实配置，并新增对 get_observability() 与 resolved_server_args_dict() 的断言，直接验证 bags 写入契约。

# test/registered/unit/observability/test_forward_pass_metrics.py
# 测试替身从手搓 SimpleNamespace 升级为真实配置发布，
# 让 reporter 走与生产相同的 get_context()/get_observability() 访问器。
​
def _publish_server_args(test, **fields):
    """Publish a config for the reporter under test and return the instance."""
    fields.setdefault("decode_log_interval", 40)
    # 通过 runtime_context 发布配置，返回的 override 可安装 / 恢复
    override = get_context().override_server_args(**fields)
    server_args = override.install()
    test.addCleanup(override.restore) # 测试结束后恢复，避免污染其他用例
    return server_args
​
​
def _make_reporter(test, scheduler) -> SchedulerMetricsReporter:
    if not hasattr(scheduler, "server_args"):
        scheduler.server_args = _publish_server_args(
            test,
            enable_metrics=False,
            enable_metrics_for_all_schedulers=False,
            kv_events_config=None,
            enable_mfu_metrics=False,
            enable_forward_pass_metrics=False,
        )
    # ... 其余 scheduler 字段填充保持不变 ...
    return SchedulerMetricsReporter(
        scheduler=scheduler,
        tp_rank=0,
        pp_rank=0,
        dp_rank=0,
        metrics_collector_context=context,
        metrics_collector=None,
    )
​
​
# 核心断言：端点由 bags 发布，实例字段不再被写入
def test_init_metrics_uses_server_worker_id(self):
    scheduler = types.SimpleNamespace()
    scheduler.server_args = _publish_server_args(
        self,
        enable_metrics=False,
        enable_metrics_for_all_schedulers=False,
        extra_metric_labels=None,
        enable_forward_pass_metrics=True,
        forward_pass_metrics_worker_id="endpoint-42",
        forward_pass_metrics_ipc_name=None,
        kv_events_config=None,
    )
    scheduler.ps = _make_ps(attn_tp_rank=0, dp_rank=2, pp_rank=0, pp_size=1)
    scheduler.enable_kv_cache_events = False
​
    with patch(
        "sglang.srt.observability.forward_pass_metrics._FpmPublisherThread",
        _DummyPublisherThread,
    ):
        reporter = _make_reporter(self, scheduler)
​
    self.assertTrue(scheduler.enable_fpm)
    self.assertEqual(scheduler._fpm_worker_id, "endpoint-42")
    self.assertEqual(scheduler._fpm_dp_rank, 2)
    self.assertEqual(scheduler._fpm_publisher.worker_id, "endpoint-42")
    self.assertEqual(scheduler._fpm_publisher.dp_rank, 2)
    self.assertTrue(scheduler._fpm_publisher.endpoint.startswith("ipc://"))
​
    # bag 写入的验证：端点可从观测命名空间与解析配置读回，
    # 但 server_args 实例上保持 None，证明不再作为消息总线。
    endpoint = get_observability().forward_pass_metrics_ipc_name
    self.assertTrue(endpoint.startswith("ipc://"))
    self.assertEqual(
        get_context().resolved_server_args_dict()["forward_pass_metrics_ipc_name"],
        endpoint,
    )
    self.assertIsNone(scheduler.server_args.forward_pass_metrics_ipc_name)

评论区精华

PR 仅 1 条来自 gemini-code-assist 机器人的评论，内容是该工具已停止 code review 服务，无实质技术讨论。PR body 提到原 #33241 的 review 讨论保留在旧 PR 上，且本 PR 代码与 #33241 最终修订版一致；当前无未解决的技术争议。

• 暂无高价值评论线程

风险与影响

• 风险：
  1. 行为回归风险（中等）：metrics_reporter.py 的 _init_fpm() 是核心初始化路径，把读/写从 server_args 实例切到 context bags。若 get_observability() 或 get_context().override() 在 reporter 初始化时尚未就绪（如测试或非标准启动流程），可能触发 AttributeError 或端点未被发布。测试已覆盖标准路径，但未覆盖 get_observability().forward_pass_metrics_ipc_name 为 None 且 override 失败的异常分支。
  2. 测试替身与真实配置的偏差（低）：测试改用真实 override_server_args 发布，若 runtime_context 的 override 机制本身有 bug，测试会跟着失败，但这也意味着测试更贴近生产。
  3. ratchet 基线刚性（低）：test_server_args_writer_ratchet.py 的 _BASELINE 是硬编码扫描计数，未来任何新增 server_args.override( 调用点都会触发失败；本 PR 只降了 1，说明该模式尚未完全清除。
     - 影响：影响范围集中在 FPM（Forward Pass Metrics）的端点发布链路和 observability 配置治理：_init_fpm() 是调度器 metrics reporter 的初始化入口，任何依赖 FPM 端点的外部消费者（如 Dynamo planner）仍可从 server_args.forward_pass_metrics_ipc_name 读回端点，但读回路径必须经由 get_context().resolved_server_args_dict()，与既有 /server_info internal_states 契约一致。对用户而言无 API 变更，但消除了 ServerArgs 实例被作为进程内消息总线使用的反模式；对团队而言，这属于持续迁移进程级配置读取/写入到命名空间访问器的一部分，与 PR #33338 的清理方向一致。影响程度中等偏低，主要惠及 observability 与配置治理的维护者。
     - 风险标记：核心初始化路径变更, 配置写入通道切换, 测试依赖真实 override 机制

关联脉络

• PR #33241 observability: publish the generated forward-pass-metrics endpoint to the bags (closed unmerged): 本 PR 直接替代 #33241，代码与 #33241 最终修订版一致；review 讨论与评论分轮处理记录均在 #33241 上。
• PR #33338 config: retire the last process-global config field reads: 同为配置治理方向：清除进程级配置读取、迁移至命名空间访问器，与本 PR 的 get_context()/get_observability() 迁移一脉相承。