From 422fa05858a93d999f6b0590d1348b6ff934747f Mon Sep 17 00:00:00 2001 From: sufubao Date: Wed, 9 Sep 2026 17:56:01 +0800 Subject: [PATCH] Add registered P/D metrics aggregation to PD master --- docs/pd_metrics.md | 77 +++++++++++++++++++++ lightllm/server/api_http_pd.py | 22 +++++- lightllm/server/metrics/pd_metrics.py | 97 +++++++++++++++++++++++++++ 3 files changed, 194 insertions(+), 2 deletions(-) create mode 100644 docs/pd_metrics.md create mode 100644 lightllm/server/metrics/pd_metrics.py diff --git a/docs/pd_metrics.md b/docs/pd_metrics.md new file mode 100644 index 0000000000..2d1f1e2684 --- /dev/null +++ b/docs/pd_metrics.md @@ -0,0 +1,77 @@ +# 通过 PD master 采集多 P、多 D 指标 + +PD master 自动从注册表获取 P/D 节点的地址，调用方无需提供 P/D 的 IP、端口或节点 ID。 + +| 接口 | 返回内容 | +| --- | --- | +| `/metrics` | master 自身指标，行为不变 | +| `/pd/metrics` | 所有已注册 P/D 节点的指标 | +| `/pd/metrics?role=prefill` | 所有已注册 P 节点的指标 | +| `/pd/metrics?role=decode` | 所有已注册 D 节点的指标 | + +```bash +curl 'http://MASTER:8000/pd/metrics' +``` + +新接口仅在 `pd_master` 模式可用，其他模式返回 404；非法 role 返回 422。 +每次请求使用注册表快照，因此节点注册、移除会自动反映到下一次采集中。 +master 必须能访问节点注册的 HTTP 地址。 + +## 指标语义与故障处理 + +各节点的指标解析后按指标族合并，保留原有样本值和标签，不对数值求和。 +每个样本增加 `pd_role="prefill|decode"` 和 `pd_node="节点注册地址"`，例如： + +```text +lightllm_num_running_reqs{pd_role="prefill",pd_node="10.0.0.1:8000",model_name="example"} 4 +lightllm_num_running_reqs{pd_role="prefill",pd_node="10.0.0.2:8000",model_name="example"} 6 +``` + +接口同时返回： + +- `lightllm_pd_scrape_success{pd_role,pd_node}`：该节点采集成功为 1，失败为 0。 +- `lightllm_pd_registered_nodes{pd_role}`：本次采集范围内各角色的注册节点数，无节点时为 0。 + +单次请求最多并发采集 16 个节点，每个节点的异步采集任务设有 5 秒超时（包含等待并发名额的时间）。 +不跟随重定向，不缓存指标。网络错误、超时、HTTP 错误、指标解析失败、重复样本、 +保留标签或指标类型冲突均会标记对应节点失败，不输出该节点的业务指标。 +`pd_role`、`pd_node` 和上述监控指标名称由 master 保留。 +节点指标过多时，解析和序列化仍会产生额外 CPU 耗时。 + +部分或全部节点失败时，接口仍返回 HTTP 200，以便 Prometheus 保存各节点的失败状态。 +因此 `up` 只表示 master 汇集接口可采集；P/D 的采集告警应使用 +`lightllm_pd_scrape_success == 0`。节点从注册表移除后不再输出其样本， +应结合注册节点数与部署期望副本数检查缺失节点。 + +## Prometheus 与 Grafana + +只需要配置 master 的地址。以下配置同时采集 master 自身与所有 P/D： + +```yaml +scrape_configs: + - job_name: lightllm-master + static_configs: + - targets: ['MASTER:8000'] + labels: + cluster: lightllm-pd + pd_role: master + + - job_name: lightllm-pd + metrics_path: /pd/metrics + scrape_interval: 15s + scrape_timeout: 10s + static_configs: + - targets: ['MASTER:8000'] + labels: + cluster: lightllm-pd +``` + +Grafana 查询 Prometheus，按 `cluster`、`pd_role`、`pd_node` 筛选。 +该配置下 `instance` 表示 master 采集入口，P/D 节点身份使用 `pd_node`。 +例如 `sum by (pd_role) (lightllm_num_running_reqs{job="lightllm-pd"})` +查看各角色的运行请求数。 +端到端请求量与延迟继续取 master 指标，避免将 master、P、D 的统计重复相加。 + +如需只采集 P 或 D，在 P/D job 中添加 `params: {role: [prefill]}` 或 +`params: {role: [decode]}`。不要同时采集全量接口和角色筛选接口，避免重复统计。 +同一组 P/D 只选择一个 master 采集入口；扩缩容不需要更新 Prometheus 的节点地址配置。 diff --git a/lightllm/server/api_http_pd.py b/lightllm/server/api_http_pd.py index 1d8b2112fc..88bd6009af 100644 --- a/lightllm/server/api_http_pd.py +++ b/lightllm/server/api_http_pd.py @@ -1,8 +1,9 @@ -"""PD separation control-plane WebSocket APIs. +"""PD separation control-plane and monitoring APIs. 供 prefill / decode 节点与 pd_master 通信： - ``/pd_register``：P/D 节点注册与请求转发 - ``/kv_move_status``：decode 节点上报 KV 传输状态 + - ``/pd/metrics``：代理已注册 P/D 节点的 Prometheus 指标 路由在模块级 ``router`` 上注册，由 ``api_http`` ``include_router`` 挂载。 ``g_objs`` 在 handler 内懒导入，避免与 api_http 循环依赖。 @@ -10,10 +11,14 @@ import asyncio import pickle +from typing import Literal, Optional import ujson as json -from fastapi import APIRouter, WebSocket, WebSocketDisconnect +from fastapi import APIRouter, HTTPException, WebSocket, WebSocketDisconnect +from fastapi.responses import Response +from prometheus_client import CONTENT_TYPE_LATEST +from lightllm.server.metrics.pd_metrics import collect_pd_metrics from lightllm.server.pd_io_struct import ObjType from lightllm.utils.envs_utils import get_lightllm_websocket_max_message_size from lightllm.utils.log_utils import init_logger @@ -23,6 +28,19 @@ router = APIRouter() +@router.get("/pd/metrics") +async def pd_metrics(role: Optional[Literal["prefill", "decode"]] = None) -> Response: + from .api_http import g_objs + + if g_objs.args.run_mode != "pd_master": + raise HTTPException(status_code=404, detail="PD metrics proxy is only available on pd_master") + + nodes = list(g_objs.httpserver_manager.pd_manager.url_to_pd_nodes.values()) + nodes = [node for node in nodes if role is None or node.mode == role] + data = await collect_pd_metrics(nodes, roles=[role] if role else ["prefill", "decode"]) + return Response(content=data, headers={"Content-Type": CONTENT_TYPE_LATEST}) + + @router.websocket("/pd_register") async def register_and_keep_alive(websocket: WebSocket): from .api_http import g_objs diff --git a/lightllm/server/metrics/pd_metrics.py b/lightllm/server/metrics/pd_metrics.py new file mode 100644 index 0000000000..cea21c5175 --- /dev/null +++ b/lightllm/server/metrics/pd_metrics.py @@ -0,0 +1,97 @@ +"""Collect registered P/D nodes without losing their individual metric series.""" + +import asyncio + +import httpx +from prometheus_client import CollectorRegistry, generate_latest +from prometheus_client.core import GaugeMetricFamily +from prometheus_client.parser import text_string_to_metric_families + +from lightllm.utils.log_utils import init_logger + +logger = init_logger(__name__) +SCRAPE_TIMEOUT = 5.0 +SCRAPE_CONCURRENCY = 16 + + +async def collect_pd_metrics(nodes, roles): + semaphore = asyncio.Semaphore(SCRAPE_CONCURRENCY) + + async def fetch(client, node): + async with semaphore: + response = await client.get(f"http://{node.client_ip_port}/metrics") + response.raise_for_status() + families = list(text_string_to_metric_families(response.text)) + if not any(family.samples for family in families): + raise ValueError("Empty metrics response") + seen = set() + for family in families: + if family.name.startswith("lightllm_pd_scrape_") or family.name == "lightllm_pd_registered_nodes": + raise ValueError("Reserved PD monitoring metric name") + for sample in family.samples: + if "pd_role" in sample.labels or "pd_node" in sample.labels: + raise ValueError("Reserved PD monitoring label") + key = (sample.name, tuple(sorted(sample.labels.items()))) + if key in seen: + raise ValueError(f"Duplicate metric sample: {sample.name}") + seen.add(key) + family.samples = [ + sample._replace(labels={**sample.labels, "pd_role": node.mode, "pd_node": node.client_ip_port}) + for sample in family.samples + ] + return families + + async with httpx.AsyncClient(timeout=SCRAPE_TIMEOUT, trust_env=False, follow_redirects=False) as client: + results = await asyncio.gather( + *(asyncio.wait_for(fetch(client, node), timeout=SCRAPE_TIMEOUT) for node in nodes), + return_exceptions=True, + ) + + merged = {} + sample_types = {} + success = GaugeMetricFamily( + "lightllm_pd_scrape_success", + "Whether a registered P/D node was scraped successfully", + labels=["pd_role", "pd_node"], + ) + for node, result in zip(nodes, results): + if not isinstance(result, BaseException): + # Validate the entire node before merging, so failed nodes contribute no partial data. + types = {name: family.type for name, family in merged.items()} + node_sample_types = dict(sample_types) + for family in result: + if (family.name in types and types[family.name] != family.type) or any( + sample.name in node_sample_types and node_sample_types[sample.name] != family.type + for sample in family.samples + ): + result = ValueError(f"Conflicting metric type: {family.name}") + break + types[family.name] = family.type + node_sample_types.update((sample.name, family.type) for sample in family.samples) + ok = not isinstance(result, BaseException) + success.add_metric([node.mode, node.client_ip_port], int(ok)) + if not ok: + logger.warning(f"P/D metrics scrape failed for {node.mode} {node.client_ip_port}: {result!r}") + continue + sample_types = node_sample_types + for family in result: + if family.name in merged: + merged[family.name].samples.extend(family.samples) + else: + merged[family.name] = family + + registered = GaugeMetricFamily( + "lightllm_pd_registered_nodes", "Number of registered P/D nodes in this scrape", labels=["pd_role"] + ) + for role in roles: + registered.add_metric([role], sum(node.mode == role for node in nodes)) + + class Collector: + def collect(self): + yield from merged.values() + yield success + yield registered + + registry = CollectorRegistry() + registry.register(Collector()) + return generate_latest(registry)