Add P/D metrics aggregation endpoint to PD master

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 的节点地址配置。

lightllm/server/api_http_pd.py

@@ -1,19 +1,24 @@
-"""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 循环依赖。
 """
 
 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

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)