Star57
188
代码
代码
Issues128
Pull Requests175
Actions
项目讨论
Wiki
分析
Star57
188
已合并
[feature]metrics格式普罗米修斯化，增强Coordinator metrics类型新增role，instance等分类类型 #136
jason lyu创建于 5月12日
[feature]metrics格式普罗米修斯化，增强Coordinator metrics类型新增role，instance等分类类型 #136
已合并
jason lyu创建于 5月12日
从codeDogPro/MindIE-PyMotor_pubmetrics_optimize合入到Ascend/MindIE-Motormaster
共 18 个文件变更+1874-1962
Mdocs/zh/api_reference/management_and_monitoring_interfaces.md+72-77
@@ -108,12 +108,12 @@ curl -X GET "http://{CoordinatorIP}:{管理端口}/readiness"
108 108 
109**接口功能**109**接口功能**
110 110 
111-返回Prometheus兼容的监控指标文本。111+返回Prometheus兼容的监控指标文本，支持通过 `type` 参数切换指标聚合粒度。
112 112 
113**接口格式**113**接口格式**
114 114 
115请求类型：**GET**115请求类型：**GET**
116-URL：`http(s)://{CoordinatorIP}:{管理端口}/metrics`116+URL：`http(s)://{CoordinatorIP}:{管理端口}/metrics?type={指标类型}&role={角色名称}`
117 117 
118 >[!NOTE]说明118 >[!NOTE]说明
119 >119 >
@@ -121,15 +121,46 @@ URL：`http(s)://{CoordinatorIP}:{管理端口}/metrics`
121 > - `{管理端口}`：配置项 `api_config.coordinator_api_mgmt_port`（默认 `1026`）。121 > - `{管理端口}`：配置项 `api_config.coordinator_api_mgmt_port`（默认 `1026`）。
122 122 
123**请求参数**123**请求参数**
124-无124+ 
125+| 参数名 | 类型 | 必选 | 默认值 | 说明 |
126+|--------|------|------|--------|------|
127+| `type` | string | 否 | `full` | 指标聚合类型：`full`（全量聚合）、`instance`（实例级）、`role`（按角色聚合） |
128+| `role` | string | 否 | 无 | 当 `type=role` 时，过滤指定角色：`prefill` 或 `decode`。不传时返回所有角色的聚合指标 |
129+ 
130+**`type` 取值说明**
131+ 
132+| 取值 | Content-Type | 返回格式 | 说明 |
133+|------|-------------|----------|------|
134+| `full`（默认） | `text/plain` | Prometheus text | 全局聚合指标，所有实例/端点的指标被聚合为单一值，可直接被 Prometheus 抓取 |
135+| `instance` | `text/plain` | Prometheus text | 实例级指标，每条指标的 label 中注入 `instance_id` 和 `role` 标签，可区分不同实例的数据 |
136+| `role`（指定 role） | `text/plain` | Prometheus text | 指定角色（`prefill` / `decode`）的聚合指标，label 中注入 `role` 标签 |
137+| `role`（不指定 role） | `text/plain` | Prometheus text | 所有角色的聚合指标拼接为单一 Prometheus 文本，可直接被 Prometheus 抓取 |
138+ 
139+**数据加工说明**
140+ 
141+Coordinator 在 `/metrics` 端点内部完成所有数据加工（实例级标签注入、角色级聚合、Prometheus 格式序列化），调用方直接获取最终格式的指标数据，无需再做二次加工。
125 142 
126**使用样例**143**使用样例**
127 144 
128```bash145```bash
146+# 全量聚合指标（默认，行为与不带参数时完全一致）
129curl -X GET "http://{CoordinatorIP}:{管理端口}/metrics"147curl -X GET "http://{CoordinatorIP}:{管理端口}/metrics"
148+curl -X GET "http://{CoordinatorIP}:{管理端口}/metrics?type=full"
149+ 
150+# 实例级指标（注入 instance_id 和 role 标签）
151+curl -X GET "http://{CoordinatorIP}:{管理端口}/metrics?type=instance"
152+ 
153+# 所有角色的聚合指标（返回 dict，key 为角色名，value 为 Prometheus 文本）
154+curl -X GET "http://{CoordinatorIP}:{管理端口}/metrics?type=role"
155+ 
156+# 仅 Prefill 角色的聚合指标
157+curl -X GET "http://{CoordinatorIP}:{管理端口}/metrics?type=role&role=prefill"
158+ 
159+# 仅 Decode 角色的聚合指标
160+curl -X GET "http://{CoordinatorIP}:{管理端口}/metrics?type=role&role=decode"
130```161```
131 162 
132-**响应示例**163+**响应示例（`type=full`，默认）**
133 164 
134```text165```text
135# HELP python_gc_objects_collected_total Objects collected during gc166# HELP python_gc_objects_collected_total Objects collected during gc
@@ -155,85 +186,50 @@ python_info{implementation="CPython",major="3",minor="11",patchlevel="10",versio
155process_virtual_memory_bytes 46601515008.0186process_virtual_memory_bytes 46601515008.0
156```187```
157 188 
158----189+**响应示例（`type=instance`）**
159 190 
160-## 实例指标查询接口191+```text
161- 192+# HELP vllm:num_requests_running Number of requests in model execution batches.
162-**接口功能**193+# TYPE vllm:num_requests_running gauge
163- 194+vllm:num_requests_running{instance_id="0",role="prefill",model_name="Qwen2.5-7B-Instruct"} 12.0
164-返回实例与端点维度的指标数据。195+vllm:num_requests_running{instance_id="1",role="prefill",model_name="Qwen2.5-7B-Instruct"} 8.0
165- 196+vllm:num_requests_running{instance_id="2",role="decode",model_name="Qwen2.5-7B-Instruct"} 6.0
166-**接口格式**197+vllm:num_requests_running{instance_id="3",role="decode",model_name="Qwen2.5-7B-Instruct"} 4.0
167- 198+# HELP vllm:kv_cache_usage_perc KV-cache usage. 1 means 100 percent usage.
168-请求类型：**GET**199+# TYPE vllm:kv_cache_usage_perc gauge
169-URL：`http(s)://{CoordinatorIP}:{管理端口}/instance/metrics`200+vllm:kv_cache_usage_perc{instance_id="0",role="prefill",model_name="Qwen2.5-7B-Instruct"} 0.62
170- 201+vllm:kv_cache_usage_perc{instance_id="1",role="prefill",model_name="Qwen2.5-7B-Instruct"} 0.45
171- >[!NOTE]说明202+vllm:kv_cache_usage_perc{instance_id="2",role="decode",model_name="Qwen2.5-7B-Instruct"} 0.72
172- >203+vllm:kv_cache_usage_perc{instance_id="3",role="decode",model_name="Qwen2.5-7B-Instruct"} 0.55
173- > - `{CoordinatorIP}`：Coordinator 服务部署机器的 IP 或域名，取值来自配置 `api_config.coordinator_api_host`（默认 `127.0.0.1`），参考 `deployer/user_config.json` 取值或实际运行时节点IP。
174- > - `{管理端口}`：配置项 `api_config.coordinator_api_mgmt_port`（默认 `1026`）。
175- 
176-**请求参数**
177-无
178- 
179-**使用样例**
180- 
181-```bash
182-curl -X GET "http://{CoordinatorIP}:{管理端口}/instance/metrics"
183```204```
184 205 
185-**响应示例**206+**响应示例（`type=role&role=prefill`）**
186 207 
187-```JSON208+```text
188-{209+# HELP vllm:num_requests_running Number of requests in model execution batches.
189- "1": [210+# TYPE vllm:num_requests_running gauge
190- {211+vllm:num_requests_running{role="prefill",model_name="Qwen2.5-7B-Instruct"} 20.0
191- "name": "process_resident_memory_bytes",212+# HELP vllm:kv_cache_usage_perc KV-cache usage. 1 means 100 percent usage.
192- "help": "Resident memory size in bytes.",213+# TYPE vllm:kv_cache_usage_perc gauge
193- "type": "gauge",214+vllm:kv_cache_usage_perc{role="prefill",model_name="Qwen2.5-7B-Instruct"} 0.535
194- "label": ["process_resident_memory_bytes"],
195- "value": [3683950592.0]
196- },
197- {
198- "name": "http_requests_total",
199- "help": "Total number of requests by method, status and handler.",
200- "type": "counter",
201- "label": [
202- "http_requests_total{handler=\"/v1/chat/completions\",method=\"POST\",status=\"2xx\"}",
203- "http_requests_total{handler=\"/v1/completions\",method=\"POST\",status=\"2xx\"}"
204- ],
205- "value": [101.0, 1.0]
206- }
207- ],
208- "2": [
209- {
210- "name": "python_gc_objects_collected_total",
211- "help": "Objects collected during gc",
212- "type": "counter",
213- "label": [
214- "python_gc_objects_collected_total{generation=\"0\"}",
215- "python_gc_objects_collected_total{generation=\"1\"}",
216- "python_gc_objects_collected_total{generation=\"2\"}"
217- ],
218- "value": [68146.0, 10867.0, 2848.0]
219- }
220- ]
221-}
222```215```
223 216 
224-**输出说明**217+**响应示例（`type=role`，不指定 role）**
225 218 
226->[!NOTE]说明219+```text
227->指标较多时仅返回结构与部分示例，具体标签维度以实际运行时为准。220+# HELP vllm:num_requests_running Number of requests in model execution batches.
228- 221+# TYPE vllm:num_requests_running gauge
229-| 参数 | 类型 | 说明 |222+vllm:num_requests_running{role="prefill",model_name="Qwen2.5-7B-Instruct"} 20.0
230-| --- | --- | --- |223+# HELP vllm:kv_cache_usage_perc KV-cache usage. 1 means 100 percent usage.
231-| <instance_id> | object | 顶层键为实例ID，值为该实例的指标列表。 |224+# TYPE vllm:kv_cache_usage_perc gauge
232-| <instance_id>[].name | string | 指标名称。 |225+vllm:kv_cache_usage_perc{role="prefill",model_name="Qwen2.5-7B-Instruct"} 0.535
233-| <instance_id>[].help | string | 指标说明。 |226+# HELP vllm:num_requests_running Number of requests in model execution batches.
234-| <instance_id>[].type | string | 指标类型，包括以下几种：<ul><li>counter</li><li>gauge</li><li>histogram</li><li>summary</li></ul>histogram和summary类型会包含 `*_bucket`、`*_count`、`*_sum` 等label项。 |227+# TYPE vllm:num_requests_running gauge
235-| <instance_id>[].label | array | 标签数组，元素为 Prometheus 标签字符串。 |228+vllm:num_requests_running{role="decode",model_name="Qwen2.5-7B-Instruct"} 10.0
236-| <instance_id>[].value | array | 与label一一对应的数值数组。 |229+# HELP vllm:kv_cache_usage_perc KV-cache usage. 1 means 100 percent usage.
230+# TYPE vllm:kv_cache_usage_perc gauge
231+vllm:kv_cache_usage_perc{role="decode",model_name="Qwen2.5-7B-Instruct"} 0.635
232+```
237 233 
238---234---
239 235 
@@ -363,7 +359,6 @@ curl -X GET "http://{CoordinatorIP}:{管理端口}/"
363 "/liveness",359 "/liveness",
364 "/readiness",360 "/readiness",
365 "/metrics",361 "/metrics",
366- "/instance/metrics",
367 "/instances/refresh"362 "/instances/refresh"
368 ],363 ],
369 "inference": [364 "inference": [
Mdocs/zh/user_guide/components/controller.md+49-0
@@ -48,6 +48,55 @@ Controller 进程入口为 `motor/controller/main.py`。启动时加载 `Control
48 48 
49另有一套 **Observability** 独立 FastAPI 应用（端口 `observability_api_port`），在 `observability_enable` 为真时提供如 inventory、metrics、alarms 等路由；未启用时相关接口返回「Observability is not enabled.」类错误（见 `observability_enabled_required`）。49另有一套 **Observability** 独立 FastAPI 应用（端口 `observability_api_port`），在 `observability_enable` 为真时提供如 inventory、metrics、alarms 等路由；未启用时相关接口返回「Observability is not enabled.」类错误（见 `observability_enabled_required`）。
50 50 
51+### Observability 应用路由
52+ 
53+| 方法 | 路径 | 说明 |
54+|------|------|------|
55+| `GET` | `/observability/inventory` | 获取服务库存信息 |
56+| `GET` | `/observability/metrics` | 获取监控指标，支持 `type` 参数切换指标粒度（见下方） |
tobking
tobkingtobking5月20日

后续对外可视化metrics统一从coordinator暴露

likedislike
jason lyu
jason lyu
5月20日 评论：
57+| `GET` | `/observability/alarms` | 获取告警信息，支持 `source_id` 查询参数过滤 |
58+ 
59+#### GET /observability/metrics
60+ 
61+> [!NOTE]路由代理说明
62+> Controller 的 `/observability/metrics` 是 Coordinator `/metrics` 的路由代理，不做数据加工。所有指标的聚合、标签注入、Prometheus 格式序列化均在 Coordinator 侧完成。Controller 仅将请求转发至 Coordinator 的 `GET /metrics?type={type}&role={role}` 并返回响应。
63+ 
64+**请求参数**
65+ 
66+| 参数名 | 类型 | 必选 | 默认值 | 说明 |
67+|--------|------|------|--------|------|
68+| `type` | string | 否 | `full` | 指标类型：`full`（全量聚合）、`instance`（实例级）、`role`（按角色聚合） |
69+| `role` | string | 否 | 无 | 当 `type=role` 时，指定角色名：`prefill` 或 `decode`。不指定时返回所有角色的聚合指标字典 |
70+ 
71+**`type` 取值说明**
72+ 
73+| 取值 | Content-Type | 返回格式 | 说明 |
74+|------|-------------|----------|------|
75+| `full` | `text/plain` | Prometheus text | 全量聚合指标（默认），Coordinator 聚合所有实例/端点后的 Prometheus 文本 |
76+| `instance` | `text/plain` | Prometheus text | 实例级指标，Coordinator 已注入 `instance_id` 和 `role` 标签 |
77+| `role`（指定 role） | `text/plain` | Prometheus text | 指定角色的聚合指标，Coordinator 已注入 `role` 标签 |
78+| `role`（不指定 role） | `text/plain` | Prometheus text | 所有角色的聚合指标拼接为单一 Prometheus 文本 |
79+ 
80+**使用样例**
81+ 
82+```bash
83+# 全量聚合指标（默认）
84+curl -X GET "http://{ControllerIP}:{observability_api_port}/observability/metrics"
85+curl -X GET "http://{ControllerIP}:{observability_api_port}/observability/metrics?type=full"
86+ 
87+# 实例级指标
88+curl -X GET "http://{ControllerIP}:{observability_api_port}/observability/metrics?type=instance"
89+ 
90+# 所有角色的聚合指标
91+curl -X GET "http://{ControllerIP}:{observability_api_port}/observability/metrics?type=role"
92+ 
93+# 仅 Prefill 角色
94+curl -X GET "http://{ControllerIP}:{observability_api_port}/observability/metrics?type=role&role=prefill"
95+ 
96+# 仅 Decode 角色
97+curl -X GET "http://{ControllerIP}:{observability_api_port}/observability/metrics?type=role&role=decode"
98+```
99+ 
51## 使用样例100## 使用样例
52 101 
53```bash102```bash
Mexamples/deployer/yaml_template/controller_template.yaml+18-2
@@ -175,5 +175,21 @@ spec:
175 app: mindie-motor-controller175 app: mindie-motor-controller
176 sessionAffinity: None176 sessionAffinity: None
177 type: ClusterIP177 type: ClusterIP
178-status:178+---
179- loadBalancer: {}179+apiVersion: v1
180+kind: Service
181+metadata:
182+ labels:
183+ app: mindie-motor-controller
184+ name: mindie-motor-controller-observability-service
185+ namespace: mindie
186+spec:
187+ ports:
188+ - nodePort: 31027
189+ port: 1027
190+ protocol: TCP
191+ targetPort: 1027
192+ selector:
193+ app: mindie-motor-controller
194+ sessionAffinity: None
195+ type: NodePort
Mexamples/deployer/yaml_template/infer_service_template.yaml+11-0
@@ -57,6 +57,17 @@ spec:
57 app: mindie-motor-controller57 app: mindie-motor-controller
58 sessionAffinity: None58 sessionAffinity: None
59 type: ClusterIP59 type: ClusterIP
60+ - name: mindie-motor-observability
61+ spec:
62+ ports:
63+ - nodePort: 31027
64+ port: 1027
65+ protocol: TCP
66+ targetPort: 1027
67+ selector:
68+ app: mindie-motor-controller
69+ sessionAffinity: None
70+ type: NodePort
60 workload:71 workload:
61 apiVersion: apps/v172 apiVersion: apps/v1
62 kind: Deployment73 kind: Deployment
Mmotor/controller/api_client/coordinator_api_client.py+25-10
@@ -35,14 +35,17 @@ class CoordinatorApiClient:
35 if event_msg.instances and len(event_msg.instances) > 0:35 if event_msg.instances and len(event_msg.instances) > 0:
36 job_names = [instance.job_name for instance in event_msg.instances]36 job_names = [instance.job_name for instance in event_msg.instances]
37 job_names_str = ", ".join(job_names)37 job_names_str = ", ".join(job_names)
38- logger.info("Event pushed type: %s, job names: [%s], response: %s",38+ logger.info(
39- event_msg.event, job_names_str, response_text)39+ "Event pushed type: %s, job names: [%s], response: %s",
40+ event_msg.event,
41+ job_names_str,
42+ response_text,
43+ )
40 else:44 else:
41- logger.info("Event pushed type: %s, push all instances, response: %s",45+ logger.info("Event pushed type: %s, push all instances, response: %s", event_msg.event, response_text)
42- event_msg.event, response_text)
43 except Exception as e:46 except Exception as e:
44 is_succeed = False47 is_succeed = False
45- address = client_ars.get('address', 'unknown')48+ address = client_ars.get("address", "unknown")
46 logger.error("Exception occurred while pushing event, %s, %s", address, e)49 logger.error("Exception occurred while pushing event, %s, %s", address, e)
47 finally:50 finally:
48 if client is not None:51 if client is not None:
@@ -62,21 +65,33 @@ class CoordinatorApiClient:
62 65 
63 @staticmethod66 @staticmethod
64 def get_full_metrics() -> str | None:67 def get_full_metrics() -> str | None:
68+ """Get full aggregated metrics from coordinator (delegates to get_metrics)."""
69+ return CoordinatorApiClient.get_metrics(metrics_type="full")
70+ 
71+ @staticmethod
72+ def get_metrics(metrics_type: str = "full", role: str | None = None) -> str | None:
65 """73 """
66- Get full aggregated metrics from coordinator (GET /metrics, plain text).74+ Get metrics from coordinator with type selection.
67- Returns None on failure.75+ Calls GET /metrics?type=<metrics_type>&role=<role>.
76+ Returns Prometheus text, or None on failure.
68 """77 """
69 client = None78 client = None
70 try:79 try:
71 client_ars = CoordinatorApiClient._generate_client_args()80 client_ars = CoordinatorApiClient._generate_client_args()
72 client = SafeHTTPSClient(**client_ars, timeout=5.0)81 client = SafeHTTPSClient(**client_ars, timeout=5.0)
73- response = client.do_get("/metrics")82+ url = f"/metrics?type={metrics_type}"
83+ if role:
84+ url += f"&role={role}"
85+ response = client.do_get(url)
74 if response and response.ok:86 if response and response.ok:
75- return response.text87+ content_type = response.headers.get("content-type", "")
88+ if "text/plain" in content_type:
89+ return response.text
90+ return response.json()
76 return None91 return None
77 except Exception as e:92 except Exception as e:
78 address = CoordinatorApiClient._generate_client_args().get("address", "unknown")93 address = CoordinatorApiClient._generate_client_args().get("address", "unknown")
79- logger.error("Failed to get full metrics from coordinator %s: %s", address, e)94+ logger.error("Failed to get metrics from coordinator %s: %s", address, e)
80 return None95 return None
81 finally:96 finally:
82 if client is not None:97 if client is not None:
Mmotor/controller/api_server/controller_api.py+76-87
@@ -1,5 +1,3 @@
1-#!/usr/bin/env python3
2-# -*- coding: utf-8 -*-
3# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.1# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.
4# MindIE is licensed under Mulan PSL v2.2# MindIE is licensed under Mulan PSL v2.
5# You can use this software according to the terms and conditions of the Mulan PSL v2.3# You can use this software according to the terms and conditions of the Mulan PSL v2.
@@ -12,14 +10,15 @@
12 10 
13import asyncio11import asyncio
14import logging12import logging
15-import os
16import threading13import threading
17from contextlib import asynccontextmanager14from contextlib import asynccontextmanager
18-from typing import Any, Callable15+from collections.abc import Callable
16+from typing import Any
19from functools import wraps17from functools import wraps
20 18 
21import uvicorn19import uvicorn
22from fastapi import FastAPI, Request, HTTPException20from fastapi import FastAPI, Request, HTTPException
21+from fastapi.responses import PlainTextResponse
23 22 
24from motor.common.resources import RegisterMsg, ReregisterMsg, HeartbeatMsg, TerminateInstanceMsg23from motor.common.resources import RegisterMsg, ReregisterMsg, HeartbeatMsg, TerminateInstanceMsg
25from motor.common.standby.standby_manager import StandbyManager, StandbyRole24from motor.common.standby.standby_manager import StandbyManager, StandbyRole
@@ -41,16 +40,20 @@ def observability_enabled_required(func: Callable):
41 @wraps(func)40 @wraps(func)
42 async def wrapper(self, *args, **kwargs):41 async def wrapper(self, *args, **kwargs):
43 if not self.enable_observability_api:42 if not self.enable_observability_api:
44- return raise_internal_error(43+ return raise_internal_error(message="Observability is not enabled.")
45- message="Observability is not enabled."
46- )
47 return await func(self, *args, **kwargs)44 return await func(self, *args, **kwargs)
45+ 
48 return wrapper46 return wrapper
49 47 
50 48 
51class ControllerAPI:49class ControllerAPI:
52- def __init__(self, config: ControllerConfig | None = None, modules: dict[str, Any] | None = None,50+ def __init__(
53- host: str = None, port: int = None):51+ self,
52+ config: ControllerConfig | None = None,
53+ modules: dict[str, Any] | None = None,
54+ host: str = None,
55+ port: int = None,
56+ ):
54 if config is None:57 if config is None:
55 config = ControllerConfig()58 config = ControllerConfig()
56 59 
@@ -82,22 +85,19 @@ class ControllerAPI:
82 85 
83 def start(self) -> None:86 def start(self) -> None:
84 # Create API server thread87 # Create API server thread
85- self.api_server_thread = threading.Thread(88+ self.api_server_thread = threading.Thread(target=self._run_api_server, daemon=True, name="APIServer")
86- target=self._run_api_server,
87- daemon=True,
88- name="APIServer"
89- )
90 self.api_server_thread.start()89 self.api_server_thread.start()
91 90 
92 # Observability API startup logic merging91 # Observability API startup logic merging
93 self.observability_api_server_thread = threading.Thread(92 self.observability_api_server_thread = threading.Thread(
94- target=self._run_observability_api_server,93+ target=self._run_observability_api_server, daemon=True, name="ObservabilityAPIServer"
95- daemon=True,
96- name="ObservabilityAPIServer"
97 )94 )
98 self.observability_api_server_thread.start()95 self.observability_api_server_thread.start()
99- logger.info("Observability API started successfully. Host: %s, Port: %d", 96+ logger.info(
100- self.observability_api_host, self.observability_api_port)97+ "Observability API started successfully. Host: %s, Port: %d",
98+ self.observability_api_host,
99+ self.observability_api_port,
100+ )
101 101 
102 logger.info("ControllerAPI started.")102 logger.info("ControllerAPI started.")
103 103 
@@ -105,10 +105,11 @@ class ControllerAPI:
105 """Check if the API server thread is alive"""105 """Check if the API server thread is alive"""
106 # Controller API server status106 # Controller API server status
107 controller_alive = self.api_server_thread is not None and self.api_server_thread.is_alive()107 controller_alive = self.api_server_thread is not None and self.api_server_thread.is_alive()
108- 108+ 
109 # check observability API server status109 # check observability API server status
110- observability_alive = (self.observability_api_server_thread is not None 110+ observability_alive = (
111- and self.observability_api_server_thread.is_alive()) 111+ self.observability_api_server_thread is not None and self.observability_api_server_thread.is_alive()
112+ )
112 return controller_alive and observability_alive113 return controller_alive and observability_alive
113 114 
114 def stop(self) -> None:115 def stop(self) -> None:
@@ -166,9 +167,17 @@ class ControllerAPI:
166 raise_internal_error(f"Internal server error: {str(e)}")167 raise_internal_error(f"Internal server error: {str(e)}")
167 168 
168 @observability_enabled_required169 @observability_enabled_required
169- async def _get_metrics(self) -> dict[str, Any]:170+ async def _get_metrics(self, request: Request):
170 try:171 try:
171- metrics_data = self.observability.get_metrics() 172+ metrics_type = request.query_params.get("type", "full").strip()
173+ role = request.query_params.get("role", None)
174+ if role is not None:
175+ role = role.strip()
176+ metrics_data = self.observability.get_metrics(metrics_type=metrics_type, role=role)
177+ # Prometheus text (full / single-role) returned as plain text;
178+ # structured data (instance / all-role) returned as JSON.
179+ if isinstance(metrics_data, str):
180+ return PlainTextResponse(content=metrics_data)
172 return format_success_response(metrics_data)181 return format_success_response(metrics_data)
173 182 
174 except Exception as e:183 except Exception as e:
@@ -178,13 +187,15 @@ class ControllerAPI:
178 @observability_enabled_required187 @observability_enabled_required
179 async def _get_alarms(self, request: Request) -> dict[str, Any]:188 async def _get_alarms(self, request: Request) -> dict[str, Any]:
180 try:189 try:
181- source_id = request.query_params.get("source_id", None) 190+ source_id = request.query_params.get("source_id", None)
182- alarms = self.observability.get_alarms(source_id=source_id) 191+ alarms = self.observability.get_alarms(source_id=source_id)
183 192 
184- return format_success_response({ 193+ return format_success_response(
185- "total": len(alarms), 194+ {
186- "alarms": alarms, 195+ "total": len(alarms),
187- })196+ "alarms": alarms,
197+ }
198+ )
188 except Exception as e:199 except Exception as e:
189 logger.error("Failed to get alarms: %s", e)200 logger.error("Failed to get alarms: %s", e)
190 raise_internal_error(f"Internal server error: {str(e)}")201 raise_internal_error(f"Internal server error: {str(e)}")
@@ -201,7 +212,7 @@ class ControllerAPI:
201 "/observability/add_alarm": logging.INFO,212 "/observability/add_alarm": logging.INFO,
202 "/startup": logging.ERROR,213 "/startup": logging.ERROR,
203 "/readiness": logging.ERROR,214 "/readiness": logging.ERROR,
204- "/liveness": logging.ERROR215+ "/liveness": logging.ERROR,
205 }216 }
206 logging.getLogger("uvicorn.access").addFilter(ApiAccessFilter(api_filters))217 logging.getLogger("uvicorn.access").addFilter(ApiAccessFilter(api_filters))
207 218 
@@ -314,23 +325,11 @@ class ControllerAPI:
314 msg = "message"325 msg = "message"
315 reason = "reason"326 reason = "reason"
316 if status.get("overall_healthy") is False:327 if status.get("overall_healthy") is False:
317- raise HTTPException(328+ raise HTTPException(status_code=503, detail={msg: "Controller is not ready", reason: "Overall not healthy"})
318- status_code=503,
319- detail={
320- msg: "Controller is not ready",
321- reason: "Overall not healthy"
322- }
323- )
324 329 
325 if status.get("deploy_mode") == "master_standby":330 if status.get("deploy_mode") == "master_standby":
326 if status.get("role") != StandbyRole.MASTER.value:331 if status.get("role") != StandbyRole.MASTER.value:
327- raise HTTPException(332+ raise HTTPException(status_code=503, detail={msg: "Controller is not ready", reason: "Not master"})
328- status_code=503,
329- detail={
330- msg: "Controller is not ready",
331- reason: "Not master"
332- }
333- )
334 return {msg: "Controller is ready"}333 return {msg: "Controller is ready"}
335 334 
336 def _get_controller_status(self) -> dict:335 def _get_controller_status(self) -> dict:
@@ -357,17 +356,13 @@ class ControllerAPI:
357 356 
358 # Check module health357 # Check module health
359 # In master_standby mode, standby node doesn't run modules, so don't check health358 # In master_standby mode, standby node doesn't run modules, so don't check health
360- if (359+ if enable_master_standby and StandbyManager.is_initialized() and not StandbyManager().is_master():
361- enable_master_standby
362- and StandbyManager.is_initialized()
363- and not StandbyManager().is_master()
364- ):
365 # Standby node: modules are not running, but this is expected360 # Standby node: modules are not running, but this is expected
366 status["overall_healthy"] = True361 status["overall_healthy"] = True
367 else:362 else:
368 unhealthy_modules = []363 unhealthy_modules = []
369 for name, module in self.modules.items():364 for name, module in self.modules.items():
370- if not hasattr(module, 'is_alive'):365+ if not hasattr(module, "is_alive"):
371 continue366 continue
372 alive = module.is_alive()367 alive = module.is_alive()
373 if not alive:368 if not alive:
@@ -392,23 +387,17 @@ class ControllerAPI:
392 # Even standby controllers should be considered alive387 # Even standby controllers should be considered alive
393 if status.get("overall_healthy") is False:388 if status.get("overall_healthy") is False:
394 raise HTTPException(389 raise HTTPException(
395- status_code=503,390+ status_code=503, detail={"message": "Controller is not alive", "reason": "Overall not healthy"}
396- detail={
397- "message": "Controller is not alive",
398- "reason": "Overall not healthy"
399- }
400 )391 )
401 else:392 else:
402 return {"message": "Controller is alive"}393 return {"message": "Controller is alive"}
403 394 
404- async def _add_alarm(self, request: Request) -> dict: 395+ async def _add_alarm(self, request: Request) -> dict:
405- body = await request.json() 396+ body = await request.json()
406- try: 397+ try:
407- if not self.enable_observability_api: 398+ if not self.enable_observability_api:
408- return format_success_response( 399+ return format_success_response(message="OM is not enabled.")
409- message="OM is not enabled." 400+ record = Record(**body)
410- )
411- record = Record(**body)
412 self.observability.add_alarm(record)401 self.observability.add_alarm(record)
413 return format_success_response()402 return format_success_response()
414 except Exception as e:403 except Exception as e:
@@ -425,8 +414,8 @@ class ControllerAPI:
425 "/observability/alarms": logging.ERROR,414 "/observability/alarms": logging.ERROR,
426 }415 }
427 logging.getLogger("uvicorn.access").addFilter(ApiAccessFilter(api_filters))416 logging.getLogger("uvicorn.access").addFilter(ApiAccessFilter(api_filters))
428- 417+ 
429- # Register middleware check role 418+ # Register middleware check role
430 app.middleware("http")(self._master_standby_middleware)419 app.middleware("http")(self._master_standby_middleware)
431 420 
432 # Register observability routes421 # Register observability routes
@@ -437,26 +426,22 @@ class ControllerAPI:
437 426 
438 return app427 return app
439 428 
440- async def _master_standby_middleware(self, request: Request, call_next): 429+ async def _master_standby_middleware(self, request: Request, call_next):
441- # if enable master/standby and is standby role then raise exception 430+ # if enable master/standby and is standby role then raise exception
442- if (431+ if self.enable_master_standby and StandbyManager.is_initialized() and not StandbyManager().is_master():
443- self.enable_master_standby432+ # raise exception at the middleware layer is an incorrect way. It is better to construct the response.
444- and StandbyManager.is_initialized()433+ raise_internal_error("This controller is not master")
445- and not StandbyManager().is_master()434+ # master continue
446- ): 435+ response = await call_next(request)
447- # raise exception at the middleware layer is an incorrect way. It is better to construct the response.
448- raise_internal_error("This controller is not master")
449- # master continue
450- response = await call_next(request)
451 return response436 return response
452 437 
453 def _run_observability_api_server(self) -> None:438 def _run_observability_api_server(self) -> None:
454 try:439 try:
455 server_config = uvicorn.Config(440 server_config = uvicorn.Config(
456- self.observability_app, 441+ self.observability_app,
457- host=self.observability_api_host, 442+ host=self.observability_api_host,
458- port=self.observability_api_port, 443+ port=self.observability_api_port,
459- log_level="info"444+ log_level="info",
460 )445 )
461 if self.observability_tls_config.enable_tls:446 if self.observability_tls_config.enable_tls:
462 server_config.load()447 server_config.load()
@@ -465,12 +450,16 @@ class ControllerAPI:
465 raise RuntimeError("Failed to create SSL context")450 raise RuntimeError("Failed to create SSL context")
466 451 
467 server_config.ssl = context452 server_config.ssl = context
468- logger.info(f"Starting observability API server on https://"453+ logger.info(
469- f"{self.observability_api_host}:{self.observability_api_port}")454+ f"Starting observability API server on https://"
455+ f"{self.observability_api_host}:{self.observability_api_port}"
456+ )
470 else:457 else:
471- logger.info(f"Starting observability API server on http://"458+ logger.info(
472- f"{self.observability_api_host}:{self.observability_api_port}")459+ f"Starting observability API server on http://"
473- 460+ f"{self.observability_api_host}:{self.observability_api_port}"
461+ )
462+ 
474 self.observability_server = uvicorn.Server(server_config)463 self.observability_server = uvicorn.Server(server_config)
475 self.observability_loop = asyncio.new_event_loop()464 self.observability_loop = asyncio.new_event_loop()
476 asyncio.set_event_loop(self.observability_loop)465 asyncio.set_event_loop(self.observability_loop)
@@ -480,4 +469,4 @@ class ControllerAPI:
480 logger.error("Observability API server error: %s", e)469 logger.error("Observability API server error: %s", e)
481 finally:470 finally:
482 if self.observability_loop and not self.observability_loop.is_closed():471 if self.observability_loop and not self.observability_loop.is_closed():
483- self.observability_loop.close()472+ self.observability_loop.close()
Mmotor/controller/observability/__init__.py+0-1
@@ -1 +0,0 @@
1-# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.
Mmotor/controller/observability/alarm/alarm_store.py+17-13
@@ -1,4 +1,3 @@
1-# -*- coding: utf-8 -*-
2# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.1# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.
3# MindIE is licensed under Mulan PSL v2.2# MindIE is licensed under Mulan PSL v2.
4# You can use this software according to the terms and conditions of the Mulan PSL v2.3# You can use this software according to the terms and conditions of the Mulan PSL v2.
@@ -11,7 +10,6 @@
11 10 
12import os11import os
13import threading12import threading
14-from typing import List
15from motor.common.alarm.record import Record13from motor.common.alarm.record import Record
16from motor.common.alarm.instance_exception_alarm import INSTANCE_EXCEPTION_ALARM_ID14from motor.common.alarm.instance_exception_alarm import INSTANCE_EXCEPTION_ALARM_ID
17from motor.common.alarm.coordinator_exception_alarm import COORDINATOR_EXCEPTION_ALARM_ID15from motor.common.alarm.coordinator_exception_alarm import COORDINATOR_EXCEPTION_ALARM_ID
@@ -25,20 +23,23 @@ logger = get_logger(__name__)
25 23 
26class AlarmStore(ThreadSafeSingleton):24class AlarmStore(ThreadSafeSingleton):
27 """Alarm manager, using thread-safe singleton pattern"""25 """Alarm manager, using thread-safe singleton pattern"""
28- 26+ 
29 def __init__(self):27 def __init__(self):
30- if hasattr(self, '_initialized'):28+ if hasattr(self, "_initialized"):
31 return29 return
32 self._initialized = True30 self._initialized = True
33 31 
34 self._dict_lock = threading.Lock()32 self._dict_lock = threading.Lock()
35- self._alarms: dict[str, List] = {os.getenv("NORTH_PLATFORM", "").strip(): []}33+ self._alarms: dict[str, list] = {os.getenv("NORTH_PLATFORM", "").strip(): []}
36 self._recoverable_alarms: dict[str, Record] = {}34 self._recoverable_alarms: dict[str, Record] = {}
37- 35+ 
38 def add_alarm(self, record: Record) -> bool:36 def add_alarm(self, record: Record) -> bool:
39 try:37 try:
40 with self._dict_lock:38 with self._dict_lock:
41- if record.alarm_id == INSTANCE_EXCEPTION_ALARM_ID or record.alarm_id == COORDINATOR_EXCEPTION_ALARM_ID:39+ if (
40+ record.alarm_id == INSTANCE_EXCEPTION_ALARM_ID
41+ or record.alarm_id == COORDINATOR_EXCEPTION_ALARM_ID
42+ ):
42 self._handle_instance_exception_alarm(record)43 self._handle_instance_exception_alarm(record)
43 else:44 else:
44 for value in self._alarms.values():45 for value in self._alarms.values():
@@ -46,7 +47,7 @@ class AlarmStore(ThreadSafeSingleton):
46 logger.debug("Current alarms: %s", self._alarms)47 logger.debug("Current alarms: %s", self._alarms)
47 48 
48 return True49 return True
49- 50+ 
50 except Exception as e:51 except Exception as e:
51 logger.error("Failed to add alarm to dict: %s", e)52 logger.error("Failed to add alarm to dict: %s", e)
52 return False53 return False
@@ -57,12 +58,15 @@ class AlarmStore(ThreadSafeSingleton):
57 result = [record.format() for record in self._alarms.get(source_id, [])]58 result = [record.format() for record in self._alarms.get(source_id, [])]
58 self._alarms[source_id] = [] # Clear alarms after fetching59 self._alarms[source_id] = [] # Clear alarms after fetching
59 return [result] if result else []60 return [result] if result else []
60- 61+ 
61-
62 def _handle_instance_exception_alarm(self, record: Record) -> None:62 def _handle_instance_exception_alarm(self, record: Record) -> None:
63 recovery_alarm_key = f"{record.alarm_id}_{record.instance_id}"63 recovery_alarm_key = f"{record.alarm_id}_{record.instance_id}"
64- logger.debug("Handling instance exception alarm with key: %s, dict keys: %s, cleared status: %s",64+ logger.debug(
65- recovery_alarm_key, list(self._recoverable_alarms.keys()), record.cleared)65+ "Handling instance exception alarm with key: %s, dict keys: %s, cleared status: %s",
66+ recovery_alarm_key,
67+ list(self._recoverable_alarms.keys()),
68+ record.cleared,
69+ )
66 if record.cleared == Cleared.NO and recovery_alarm_key not in self._recoverable_alarms:70 if record.cleared == Cleared.NO and recovery_alarm_key not in self._recoverable_alarms:
67 for value in self._alarms.values():71 for value in self._alarms.values():
68 value.append(record)72 value.append(record)
@@ -70,4 +74,4 @@ class AlarmStore(ThreadSafeSingleton):
70 if record.cleared == Cleared.YES and recovery_alarm_key in self._recoverable_alarms:74 if record.cleared == Cleared.YES and recovery_alarm_key in self._recoverable_alarms:
71 for value in self._alarms.values():75 for value in self._alarms.values():
72 value.append(record)76 value.append(record)
73- del self._recoverable_alarms[recovery_alarm_key]77+ del self._recoverable_alarms[recovery_alarm_key]
Mmotor/controller/observability/inventory/inventory_collector.py+39-41
@@ -1,4 +1,3 @@
1-# -*- coding: utf-8 -*-
2# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.1# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.
3# MindIE is licensed under Mulan PSL v2.2# MindIE is licensed under Mulan PSL v2.
4# You can use this software according to the terms and conditions of the Mulan PSL v2.3# You can use this software according to the terms and conditions of the Mulan PSL v2.
@@ -80,8 +79,7 @@ def _get_instance_info_from_instance(instance: Instance, instance_status: Instan
80 pod_associated_info_list = []79 pod_associated_info_list = []
81 for _, endpoint_info in pod_info.items():80 for _, endpoint_info in pod_info.items():
82 for device_info in endpoint_info.device_infos:81 for device_info in endpoint_info.device_infos:
83- new_pod_info = {NPU_ID: device_info.device_id,82+ new_pod_info = {NPU_ID: device_info.device_id, NPU_IP: device_info.device_ip}
84- NPU_IP: device_info.device_ip}
85 pod_associated_info_list.append(new_pod_info)83 pod_associated_info_list.append(new_pod_info)
86 84 
87 new_pod_info = {POD_ASSOCIATED_INFO_LIST: pod_associated_info_list, POD_ID: pod_ip, POD_NAME: ""}85 new_pod_info = {POD_ASSOCIATED_INFO_LIST: pod_associated_info_list, POD_ID: pod_ip, POD_NAME: ""}
@@ -93,7 +91,7 @@ def _get_instance_info_from_instance(instance: Instance, instance_status: Instan
93 "InstanceStatus": instance_status,91 "InstanceStatus": instance_status,
94 "podInfoList": pod_info_list,92 "podInfoList": pod_info_list,
95 SERVER_IP_LIST: list(set(server_ip_list)),93 SERVER_IP_LIST: list(set(server_ip_list)),
96- "serverList": []94+ "serverList": [],
97 }95 }
98 return instance_info96 return instance_info
99 97 
@@ -110,13 +108,15 @@ class InventoryCollector(ThreadSafeSingleton):
110 self.initial_instance_list = InstanceManager().get_initial_instances()108 self.initial_instance_list = InstanceManager().get_initial_instances()
111 self.inactive_instance_list = InstanceManager().get_inactive_instances()109 self.inactive_instance_list = InstanceManager().get_inactive_instances()
112 model_id = os.getenv(MODEL_ID, "")110 model_id = os.getenv(MODEL_ID, "")
113- model_service_info = {"inventories": self._collect_inventory_detail(),111+ model_service_info = {
114- "inferenceFrameworkType": f"motor-{os.getenv('ENGINE_TYPE', '').lower()}",112+ "inventories": self._collect_inventory_detail(),
115- "modelID": model_id,113+ "inferenceFrameworkType": f"motor-{os.getenv('ENGINE_TYPE', '').lower()}",
116- "modelName": self._get_model_name(),114+ "modelID": model_id,
117- "modelState": self._get_model_state(),115+ "modelName": self._get_model_name(),
118- "modelType": self._get_model_name(),116+ "modelState": self._get_model_state(),
119- "timestamp": int(time.time() * 1000)}117+ "modelType": self._get_model_name(),
118+ "timestamp": int(time.time() * 1000),
119+ }
120 return model_service_info120 return model_service_info
121 121 
122 def _collect_inventory_detail(self) -> dict:122 def _collect_inventory_detail(self) -> dict:
@@ -127,17 +127,18 @@ class InventoryCollector(ThreadSafeSingleton):
127 for instance_info in d_instance_list:127 for instance_info in d_instance_list:
128 all_server_ip_list.extend(instance_info.get(SERVER_IP_LIST))128 all_server_ip_list.extend(instance_info.get(SERVER_IP_LIST))
129 129 
130- inventory_detail = {"DInstanceList": d_instance_list,130+ inventory_detail = {
131- "PInstanceList": p_instance_list,131+ "DInstanceList": d_instance_list,
132- "DPGroupList": self._get_dp_group_list(),132+ "PInstanceList": p_instance_list,
133- "PDHybridList": [],133+ "DPGroupList": self._get_dp_group_list(),
134- "backupServerList": _get_backup_server_list(),134+ "PDHybridList": [],
135- "expertList": _get_expert_list(),135+ "backupServerList": _get_backup_server_list(),
136- SERVER_IP_LIST: list(set(all_server_ip_list)),136+ "expertList": _get_expert_list(),
137- "serverOfCoordinator": _get_server_of_coordinator(),137+ SERVER_IP_LIST: list(set(all_server_ip_list)),
138- "serverOfManagerMaster": _get_server_of_controller_master(),138+ "serverOfCoordinator": _get_server_of_coordinator(),
139- "serverOfManagerSlave": _get_server_of_controller_slave()139+ "serverOfManagerMaster": _get_server_of_controller_master(),
140- }140+ "serverOfManagerSlave": _get_server_of_controller_slave(),
141+ }
141 return inventory_detail142 return inventory_detail
142 143 
143 def _get_dp_group_list(self):144 def _get_dp_group_list(self):
@@ -158,28 +159,25 @@ class InventoryCollector(ThreadSafeSingleton):
158 # every endpoint_info equals to one dp159 # every endpoint_info equals to one dp
159 npu_info_list = []160 npu_info_list = []
160 for device_info in endpoint_info.device_infos:161 for device_info in endpoint_info.device_infos:
161- npu_info = {NPU_ID: device_info.device_id,162+ npu_info = {NPU_ID: device_info.device_id, NPU_IP: device_info.device_ip}
162- NPU_IP: device_info.device_ip
163- }
164 npu_info_list.append(npu_info)163 npu_info_list.append(npu_info)
165 pod_npu_info_list.extend(npu_info_list)164 pod_npu_info_list.extend(npu_info_list)
166- server_info = {"NPUInfoList": npu_info_list,165+ server_info = {
167- "serverID": host_ip, "serverIP": pod_ip, "serverName": ""166+ "NPUInfoList": npu_info_list,
168- }167+ "serverID": host_ip,
169- new_pod_info = {POD_ASSOCIATED_INFO_LIST: [],168+ "serverIP": pod_ip,
170- POD_ID: pod_ip,169+ "serverName": "",
171- POD_NAME: ""
172- }
173- dp_info = {"DPID": endpoint_id, "DPName": "",
174- "DPRole": DPRole.CENTRAL, "PDInstID": temp_instance_name,
175- "podInfoList": [new_pod_info],
176- "serverList": [server_info]
177- }
178- dp_group = {
179- "DPGroupID": endpoint_id,
180- "DPGroupName": endpoint_id,
181- "DPList": [dp_info]
182 }170 }
171+ new_pod_info = {POD_ASSOCIATED_INFO_LIST: [], POD_ID: pod_ip, POD_NAME: ""}
172+ dp_info = {
173+ "DPID": endpoint_id,
174+ "DPName": "",
175+ "DPRole": DPRole.CENTRAL,
176+ "PDInstID": temp_instance_name,
177+ "podInfoList": [new_pod_info],
178+ "serverList": [server_info],
179+ }
180+ dp_group = {"DPGroupID": endpoint_id, "DPGroupName": endpoint_id, "DPList": [dp_info]}
183 temp_dp_group_list.append(dp_group)181 temp_dp_group_list.append(dp_group)
184 for dp_group in temp_dp_group_list:182 for dp_group in temp_dp_group_list:
185 for dp_info in dp_group["DPList"]:183 for dp_info in dp_group["DPList"]:
Mmotor/controller/observability/metrics/metrics_collector.py+16-42
@@ -1,4 +1,3 @@
1-# -*- coding: utf-8 -*-
2# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.1# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.
3# MindIE is licensed under Mulan PSL v2.2# MindIE is licensed under Mulan PSL v2.
4# You can use this software according to the terms and conditions of the Mulan PSL v2.3# You can use this software according to the terms and conditions of the Mulan PSL v2.
@@ -10,57 +9,32 @@
10# See the Mulan PSL v2 for more details.9# See the Mulan PSL v2 for more details.
11 10 
12"""11"""
13-Controller OM metrics collector: client to coordinator for full (aggregated) metrics.12+Controller Observability metrics collector: thin proxy that routes metrics
14-If the interval between two collections is less than 15 seconds, returns the last13+requests to the Coordinator. All data processing (aggregation, label injection,
15-collected metrics; otherwise fetches from coordinator. Only full metrics are supported,14+Prometheus serialization) is handled by the Coordinator.
16-not per-instance metrics.
17"""15"""
18 16 
19-import threading
20-import time
21-from typing import Optional
22- 
23-from motor.common.logger import get_logger
24from motor.controller.api_client.coordinator_api_client import CoordinatorApiClient17from motor.controller.api_client.coordinator_api_client import CoordinatorApiClient
25-from motor.config.controller import ControllerConfig
26- 
27-logger = get_logger(__name__)
28 18 
29 19 
30class MetricsCollector:20class MetricsCollector:
31 """21 """
32- Client-side metrics collector: fetches full metrics from coordinator with 15s cache.22+ Client-side metrics collector: routes requests to Coordinator.
33 """23 """
34 24 
35- def __init__(self, config: ControllerConfig | None = None) -> None:25+ def __init__(self) -> None:
36- if config is None:26+ pass
37- config = ControllerConfig()
38- self.config = config
39- 
40- self._lock = threading.RLock()
41- self._last_metrics: Optional[str] = None
42- self._last_fetch_time: float = 0.0
43- self._cache_ttl_sec = self.config.observability_config.metrics_ttl
44 27 
45 def get_full_metrics(self) -> str:28 def get_full_metrics(self) -> str:
46- """29+ result = CoordinatorApiClient.get_metrics(metrics_type="full")
47- Get full aggregated metrics. If last fetch was within 15 seconds, returns30+ return result if result is not None else ""
48- cached value; otherwise fetches from coordinator and updates cache.
49- Returns empty string on error (and keeps previous cache for next call).
50- """
51- now = time.monotonic()
52- with self._lock:
53- if (self._last_metrics is not None and
54- (now - self._last_fetch_time) < self._cache_ttl_sec):
55- return self._last_metrics
56 31 
57- metrics = CoordinatorApiClient.get_full_metrics()32+ def get_instance_metrics_prometheus(self) -> str:
58- with self._lock:33+ result = CoordinatorApiClient.get_metrics(metrics_type="instance")
59- if metrics is not None:34+ return result if result is not None else ""
60- self._last_metrics = metrics35+ 
61- self._last_fetch_time = time.monotonic()36+ def get_role_metrics(self, role: str | None = None) -> str:
62- return self._last_metrics37+ result = CoordinatorApiClient.get_metrics(metrics_type="role", role=role)
63- # On failure, return last cached if any38+ if result is None:
64- if self._last_metrics is not None:
65- return self._last_metrics
66 return ""39 return ""
40+ return result
Mmotor/controller/observability/observability.py+22-19
@@ -1,4 +1,3 @@
1-# -*- coding: utf-8 -*-
2# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.1# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.
3# MindIE is licensed under Mulan PSL v2.2# MindIE is licensed under Mulan PSL v2.
4# You can use this software according to the terms and conditions of the Mulan PSL v2.3# You can use this software according to the terms and conditions of the Mulan PSL v2.
@@ -14,7 +13,7 @@ from motor.common.logger import get_logger
14from motor.common.utils.singleton import ThreadSafeSingleton13from motor.common.utils.singleton import ThreadSafeSingleton
15from motor.config.controller import ControllerConfig14from motor.config.controller import ControllerConfig
16from motor.controller.observability.alarm.alarm_store import AlarmStore15from motor.controller.observability.alarm.alarm_store import AlarmStore
17-from motor.controller.observability.metrics.metrics_collector import MetricsCollector16+from motor.controller.api_client.coordinator_api_client import CoordinatorApiClient
18 17 
19 18 
20logger = get_logger(__name__)19logger = get_logger(__name__)
@@ -22,15 +21,14 @@ logger = get_logger(__name__)
22 21 
23class Observability(ThreadSafeSingleton):22class Observability(ThreadSafeSingleton):
24 """23 """
25- Observability24+ Observability module for Motor Controller: manages alarms and metrics routing.
26- unified management and coordination of all operation and maintenance modules
27- 
28 """25 """
26+ 
29 def __init__(self, config: ControllerConfig | None = None) -> None:27 def __init__(self, config: ControllerConfig | None = None) -> None:
30 super().__init__()28 super().__init__()
31- 29+ 
32- # If already initialized, return 30+ # If already initialized, return
33- if hasattr(self, '_initialized'):31+ if hasattr(self, "_initialized"):
34 return32 return
35 self._initialized = True33 self._initialized = True
36 34 
@@ -39,20 +37,15 @@ class Observability(ThreadSafeSingleton):
39 self.config = config37 self.config = config
40 38 
41 self.alarm_store = AlarmStore()39 self.alarm_store = AlarmStore()
42- self.metrics_collector = MetricsCollector(self.config)40+ 
43-
44 def start(self) -> None:41 def start(self) -> None:
45- """start observability"""
46 logger.info("Starting observability...")42 logger.info("Starting observability...")
47 43 
48 def stop(self) -> None:44 def stop(self) -> None:
49- """stop observability"""
50 logger.info("Stopping observability...")45 logger.info("Stopping observability...")
51- 46+ 
52 def add_alarm(self, record: Record) -> bool:47 def add_alarm(self, record: Record) -> bool:
53 try:48 try:
54- from motor.common.alarm.instance_exception_alarm import InstanceExceptionAlarm
55-
56 is_succeed = self.alarm_store.add_alarm(record)49 is_succeed = self.alarm_store.add_alarm(record)
57 if is_succeed:50 if is_succeed:
58 logger.debug("Alarm added successfully via observability, %s", str(record))51 logger.debug("Alarm added successfully via observability, %s", str(record))
@@ -70,9 +63,19 @@ class Observability(ThreadSafeSingleton):
70 logger.error("Failed to get alarms via observability: %s", e)63 logger.error("Failed to get alarms via observability: %s", e)
71 return []64 return []
72 65 
73- def get_metrics(self) -> str:66+ def get_metrics(self, metrics_type: str = "full", role: str | None = None) -> str:
67+ """
68+ Get metrics by type. Routes request to Coordinator for data processing.
69+ 
70+ :param metrics_type: "full" (default), "instance", or "role"
71+ :param role: when metrics_type is "role", filter to a specific role (e.g. "prefill", "decode")
72+ :returns: Prometheus text
73+ """
74 try:74 try:
75- return self.metrics_collector.get_full_metrics()75+ result = CoordinatorApiClient.get_metrics(metrics_type=metrics_type, role=role)
76+ if result is None:
77+ return ""
78+ return result
76 except Exception as e:79 except Exception as e:
77- logger.error(f"Failed to get metrics via observability: {e}")80+ logger.error("Failed to get metrics via observability: %s", e)
78- return ""81+ return ""
Mmotor/coordinator/api_server/management_server.py+12-21
@@ -43,6 +43,7 @@ from motor.coordinator.domain.probe import (
43 ReadinessResult,43 ReadinessResult,
44 RoleShmDaemonLivenessProvider,44 RoleShmDaemonLivenessProvider,
45)45)
46+ 
46logger = get_logger(__name__)47logger = get_logger(__name__)
47 48 
48# Readiness 503: result -> HTTP detail.49# Readiness 503: result -> HTTP detail.
@@ -90,8 +91,7 @@ class ManagementServer(BaseCoordinatorServer):
90 # Create dependencies before app so lifespan and routes see them (lifespan runs on uvicorn start)91 # Create dependencies before app so lifespan and routes see them (lifespan runs on uvicorn start)
91 self._scheduler_connection = SchedulerConnectionManager.from_config(self.coordinator_config)92 self._scheduler_connection = SchedulerConnectionManager.from_config(self.coordinator_config)
92 self._instance_manager = (93 self._instance_manager = (
93- instance_manager if instance_manager is not None94+ instance_manager if instance_manager is not None else InstanceManager(self.coordinator_config, TYPE_MGMT)
94- else InstanceManager(self.coordinator_config, TYPE_MGMT)
95 )95 )
96 deploy_mode = (96 deploy_mode = (
97 self.coordinator_config.scheduler_config.deploy_mode97 self.coordinator_config.scheduler_config.deploy_mode
@@ -237,11 +237,7 @@ class ManagementServer(BaseCoordinatorServer):
237 status_code=status.HTTP_503_SERVICE_UNAVAILABLE,237 status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
238 detail=_READINESS_503[out.result],238 detail=_READINESS_503[out.result],
239 )239 )
240- msg = (240+ msg = "Coordinator is master" if out.result == ReadinessResult.OK_MASTER else "Coordinator is ok"
241- "Coordinator is master"
242- if out.result == ReadinessResult.OK_MASTER
243- else "Coordinator is ok"
244- )
245 logger.info(241 logger.info(
246 "[Readiness] is_ready=%s instances_status=%s -> 200",242 "[Readiness] is_ready=%s instances_status=%s -> 200",
247 out.is_ready,243 out.is_ready,
@@ -250,13 +246,13 @@ class ManagementServer(BaseCoordinatorServer):
250 return _build_readiness_response(msg, out.is_ready)246 return _build_readiness_response(msg, out.is_ready)
251 247 
252 @self.management_app.get("/metrics")248 @self.management_app.get("/metrics")
253- async def get_metrics():249+ async def get_metrics(request: Request):
254- metrics = MetricsCollector().prometheus_metrics_handler()250+ metrics_type = request.query_params.get("type", "full")
255- return PlainTextResponse(content=metrics)251+ role = request.query_params.get("role", None)
256- 252+ result = MetricsCollector().get_metrics(metrics_type=metrics_type, role=role)
257- @self.management_app.get("/instance/metrics")253+ if isinstance(result, str):
258- async def get_instance_metrics():254+ return PlainTextResponse(content=result)
259- return MetricsCollector().prometheus_instance_metrics_handler()255+ return result
260 256 
261 @self.management_app.post("/instances/refresh", response_model=RequestResponse)257 @self.management_app.post("/instances/refresh", response_model=RequestResponse)
262 @self.timeout_handler()258 @self.timeout_handler()
@@ -290,7 +286,6 @@ class ManagementServer(BaseCoordinatorServer):
290 "GET /startup": "startup probe",286 "GET /startup": "startup probe",
291 "GET /readiness": "readiness check",287 "GET /readiness": "readiness check",
292 "GET /metrics": "get metrics",288 "GET /metrics": "get metrics",
293- "GET /instance/metrics": "get instance metrics",
294 "POST /instances/refresh": "refresh instances",289 "POST /instances/refresh": "refresh instances",
295 },290 },
296 }291 }
@@ -308,8 +303,7 @@ class ManagementServer(BaseCoordinatorServer):
308 raise HTTPException(303 raise HTTPException(
309 status_code=status.HTTP_400_BAD_REQUEST,304 status_code=status.HTTP_400_BAD_REQUEST,
310 detail=(305 detail=(
311- "Request body size exceeds maximum allowed size of "306+ f"Request body size exceeds maximum allowed size of {_MAX_REQUEST_BODY_SIZE // (1024 * 1024)}MB"
312- f"{_MAX_REQUEST_BODY_SIZE // (1024 * 1024)}MB"
313 ),307 ),
314 )308 )
315 body = json.loads(raw_body.decode("utf-8"))309 body = json.loads(raw_body.decode("utf-8"))
@@ -322,10 +316,7 @@ class ManagementServer(BaseCoordinatorServer):
322 raise316 raise
323 except json.JSONDecodeError as e:317 except json.JSONDecodeError as e:
324 logger.error("Failed to parse request body as JSON: %s", e)318 logger.error("Failed to parse request body as JSON: %s", e)
325- preview = (319+ preview = raw_body.decode("utf-8", errors="ignore")[:_REQUEST_BODY_PREVIEW_LENGTH] if raw_body else "empty"
326- raw_body.decode("utf-8", errors="ignore")[:_REQUEST_BODY_PREVIEW_LENGTH]
327- if raw_body else "empty"
328- )
329 logger.error("Request body (first %s chars): %s", _REQUEST_BODY_PREVIEW_LENGTH, preview)320 logger.error("Request body (first %s chars): %s", _REQUEST_BODY_PREVIEW_LENGTH, preview)
330 raise HTTPException(321 raise HTTPException(
331 status_code=status.HTTP_400_BAD_REQUEST,322 status_code=status.HTTP_400_BAD_REQUEST,
Mtests/controller/api_server/test_controller_api.py+202-123
@@ -1,5 +1,3 @@
1-#!/usr/bin/env python3
2-# -*- coding: utf-8 -*-
3# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.1# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.
4# MindIE is licensed under Mulan PSL v2.2# MindIE is licensed under Mulan PSL v2.
5# You can use this software according to the terms and conditions of the Mulan PSL v2.3# You can use this software according to the terms and conditions of the Mulan PSL v2.
Mtests/controller/observability/metrics/test_metrics_collector.py+69-236
@@ -8,264 +8,97 @@
8# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.8# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
9# See the Mulan PSL v2 for more details.9# See the Mulan PSL v2 for more details.
10 10 
11-import threading
12-import time
13-from unittest.mock import patch, Mock
14import pytest11import pytest
12+from unittest.mock import patch
15 13 
16-from motor.common.logger import get_logger
17-from motor.controller.api_client.coordinator_api_client import CoordinatorApiClient
18-from motor.config.controller import ControllerConfig
19from motor.controller.observability.metrics.metrics_collector import MetricsCollector14from motor.controller.observability.metrics.metrics_collector import MetricsCollector
20 15 
21-# Define test metrics data
22-TEST_METRICS_DATA = """# HELP vllm:request_success_total Count of successfully processed requests.
23-# TYPE vllm:request_success_total counter
24-vllm:request_success_total{engine="0",finished_reason="stop",model_name="/job/model/Qwen2.5-0.5B-Instruct"} 1.0
25-vllm:request_success_total{engine="0",finished_reason="length",model_name="/job/model/Qwen2.5-0.5B-Instruct"} 2.0
26-vllm:request_success_total{engine="0",finished_reason="abort",model_name="/job/model/Qwen2.5-0.5B-Instruct"} 0.0"""
27 16 
28@pytest.fixture17@pytest.fixture
29def metrics_collector():18def metrics_collector():
30 """Fixture to create a MetricsCollector instance"""19 """Fixture to create a MetricsCollector instance"""
31- config = ControllerConfig()20+ return MetricsCollector()
32- config.observability_config.metrics_ttl = 2 # Set a short TTL for testing
33- collector = MetricsCollector(config)
34- return collector
35 21 
36 22 
37-@patch("motor.controller.observability.metrics.metrics_collector.time.monotonic")23+@patch("motor.controller.observability.metrics.metrics_collector.CoordinatorApiClient.get_metrics")
38-@patch("motor.controller.observability.metrics.metrics_collector.CoordinatorApiClient.get_full_metrics")24+def test_get_full_metrics(mock_get_metrics, metrics_collector):
39-def test_get_full_metrics_normal_case(mock_get_metrics, mock_monotonic, metrics_collector):25+ mock_get_metrics.return_value = "# HELP test test_metric\ntest_metric 1.0"
40- """26+ 
41- Test normal case: First-time metrics fetch
42- """
43- # Prepare test data
44- mock_get_metrics.return_value = TEST_METRICS_DATA
45-
46- # Mock time
47- mock_monotonic.return_value = 1000.0
48-
49- # Execute
50 result = metrics_collector.get_full_metrics()27 result = metrics_collector.get_full_metrics()
51- 28+ 
52- # Verify29+ assert result == "# HELP test test_metric\ntest_metric 1.0"
53- assert result == TEST_METRICS_DATA30+ mock_get_metrics.assert_called_once_with(metrics_type="full")
54- mock_get_metrics.assert_called_once()
55-
56- # Verify cache
57- with metrics_collector._lock:
58- assert metrics_collector._last_metrics == TEST_METRICS_DATA
59- assert metrics_collector._last_fetch_time == 1000.0
60 31 
61 32 
62-@patch("motor.controller.observability.metrics.metrics_collector.time.monotonic")33+@patch("motor.controller.observability.metrics.metrics_collector.CoordinatorApiClient.get_metrics")
63-@patch("motor.controller.observability.metrics.metrics_collector.CoordinatorApiClient.get_full_metrics")34+def test_get_full_metrics_api_failure(mock_get_metrics, metrics_collector):
64-def test_get_full_metrics_cache_hit(mock_get_metrics, mock_monotonic, metrics_collector):
65- """
66- Test cache hit: Return cached data within TTL
67- """
68- # First call, set cache
69- mock_get_metrics.return_value = TEST_METRICS_DATA
70- mock_monotonic.return_value = 1000.0
71- result1 = metrics_collector.get_full_metrics()
72-
73- assert result1 == TEST_METRICS_DATA
74- mock_get_metrics.assert_called_once()
75-
76- # Second call, within TTL, should hit cache
77- mock_get_metrics.reset_mock()
78- mock_monotonic.return_value = 1001.9 # Still within TTL
79- result2 = metrics_collector.get_full_metrics()
80-
81- # Verify: Should return cache, and API is not called again
82- assert result2 == TEST_METRICS_DATA
83- mock_get_metrics.assert_not_called()
84- 
85- 
86-@patch("motor.controller.observability.metrics.metrics_collector.time.monotonic")
87-@patch("motor.controller.observability.metrics.metrics_collector.CoordinatorApiClient.get_full_metrics")
88-def test_get_full_metrics_cache_expired(mock_get_metrics, mock_monotonic, metrics_collector):
89- """
90- Test cache expiration: Refetch after TTL
91- """
92- # Prepare new metrics data
93- NEW_METRICS_DATA = TEST_METRICS_DATA + """
94-vllm:request_success_total{engine="1",finished_reason="stop",model_name="/job/model/Qwen2.5-0.5B-Instruct"} 5.0"""
95-
96- # First call, set cache
97- mock_get_metrics.return_value = TEST_METRICS_DATA
98- mock_monotonic.return_value = 1000.0
99- result1 = metrics_collector.get_full_metrics()
100-
101- # Second call, TTL expired, should refetch
102- mock_get_metrics.reset_mock()
103- mock_get_metrics.return_value = NEW_METRICS_DATA
104- mock_monotonic.return_value = 1002.1 # Exceeds 2-second TTL
105- result2 = metrics_collector.get_full_metrics()
106-
107- # Verify: Should return new metrics
108- assert result2 == NEW_METRICS_DATA
109- mock_get_metrics.assert_called_once()
110-
111- # Verify cache is updated
112- with metrics_collector._lock:
113- assert metrics_collector._last_metrics == NEW_METRICS_DATA
114- assert metrics_collector._last_fetch_time == 1002.1
115- 
116- 
117-@patch("motor.controller.observability.metrics.metrics_collector.time.monotonic")
118-@patch("motor.controller.observability.metrics.metrics_collector.CoordinatorApiClient.get_full_metrics")
119-def test_get_full_metrics_api_failure_with_cache(mock_get_metrics, mock_monotonic, metrics_collector):
120- """
121- Test API failure with existing cache
122- """
123- # First successful call, set cache
124- mock_get_metrics.return_value = TEST_METRICS_DATA
125- mock_monotonic.return_value = 1000.0
126- result1 = metrics_collector.get_full_metrics()
127-
128- assert result1 == TEST_METRICS_DATA
129-
130- # Second call, cache expired but API call fails
131- mock_get_metrics.reset_mock()
132- mock_get_metrics.return_value = None # API call fails
133- mock_monotonic.return_value = 1002.5 # Exceeds TTL
134-
135- result2 = metrics_collector.get_full_metrics()
136-
137- # Verify: Should return old cache
138- assert result2 == TEST_METRICS_DATA
139- mock_get_metrics.assert_called_once()
140-
141- # Verify fetch time is not updated
142- with metrics_collector._lock:
143- assert metrics_collector._last_fetch_time == 1000.0
144- 
145- 
146-@patch("motor.controller.observability.metrics.metrics_collector.time.monotonic")
147-@patch("motor.controller.observability.metrics.metrics_collector.CoordinatorApiClient.get_full_metrics")
148-def test_get_full_metrics_api_failure_without_cache(mock_get_metrics, mock_monotonic, metrics_collector):
149- """
150- Test API failure without cache
151- """
152- # First call fails immediately
153 mock_get_metrics.return_value = None35 mock_get_metrics.return_value = None
154- mock_monotonic.return_value = 1000.036+ 
155-
156 result = metrics_collector.get_full_metrics()37 result = metrics_collector.get_full_metrics()
157- 38+ 
158- # Verify: Should return empty string
159 assert result == ""39 assert result == ""
160- mock_get_metrics.assert_called_once()40+ mock_get_metrics.assert_called_once_with(metrics_type="full")
161-
162- # Verify cache is empty
163- with metrics_collector._lock:
164- assert metrics_collector._last_metrics is None
165- assert metrics_collector._last_fetch_time == 0.0
166 41 
167 42 
168-@patch("motor.controller.observability.metrics.metrics_collector.time.monotonic")43+@patch("motor.controller.observability.metrics.metrics_collector.CoordinatorApiClient.get_metrics")
169-@patch("motor.controller.observability.metrics.metrics_collector.CoordinatorApiClient.get_full_metrics")44+def test_get_instance_metrics_prometheus(mock_get_metrics, metrics_collector):
170-def test_get_full_metrics_concurrent_access(mock_get_metrics, mock_monotonic, metrics_collector):45+ mock_get_metrics.return_value = (
171- """46+ '# HELP test test_metric\n'
172- Test concurrent access: Multiple threads accessing simultaneously47+ '# TYPE test_metric gauge\n'
173- """48+ 'test_metric{instance_id="1",role="prefill"} 5.0'
174- import threading49+ )
175- 50+ 
176- # Record API call count51+ result = metrics_collector.get_instance_metrics_prometheus()
177- call_count = 052+ 
178- 53+ assert "instance_id" in result
179- def call_api():54+ assert "prefill" in result
180- nonlocal call_count55+ mock_get_metrics.assert_called_once_with(metrics_type="instance")
181- call_count += 1
182- return TEST_METRICS_DATA
183-
184- mock_get_metrics.side_effect = call_api
185-
186- # Set time points to simulate cache expiration
187- time_points = iter([1000.0, 1001.0, 1001.1, 1001.2])
188- mock_monotonic.side_effect = lambda: next(time_points)
189-
190- # Create multiple threads for concurrent calls
191- results = []
192- threads = []
193-
194- def worker():
195- result = metrics_collector.get_full_metrics()
196- results.append(result)
197-
198- for _ in range(3):
199- t = threading.Thread(target=worker)
200- threads.append(t)
201- t.start()
202-
203- for t in threads:
204- t.join()
205-
206- # Verify: API should be called only once (due to lock)
207- assert call_count == 1
208- assert len(results) == 3
209- assert all(r == TEST_METRICS_DATA for r in results)
210 56 
211 57 
212-@patch("motor.controller.observability.metrics.metrics_collector.time.monotonic")58+@patch("motor.controller.observability.metrics.metrics_collector.CoordinatorApiClient.get_metrics")
213-@patch("motor.controller.observability.metrics.metrics_collector.CoordinatorApiClient.get_full_metrics")59+def test_get_instance_metrics_prometheus_api_failure(mock_get_metrics, metrics_collector):
214-def test_get_full_metrics_empty_string_response(mock_get_metrics, mock_monotonic, metrics_collector):60+ mock_get_metrics.return_value = None
215- """61+ 
216- Test API returning empty string62+ result = metrics_collector.get_instance_metrics_prometheus()
217- """63+ 
218- # API returns empty string
219- mock_get_metrics.return_value = ""
220- mock_monotonic.return_value = 1000.0
221-
222- result = metrics_collector.get_full_metrics()
223-
224- # Verify: Should return empty string
225 assert result == ""64 assert result == ""
226- mock_get_metrics.assert_called_once()65+ mock_get_metrics.assert_called_once_with(metrics_type="instance")
227-
228- # Verify cache is set to empty string
229- with metrics_collector._lock:
230- assert metrics_collector._last_metrics == ""
231 66 
232 67 
233-def test_metrics_collector_default_config():68+@patch("motor.controller.observability.metrics.metrics_collector.CoordinatorApiClient.get_metrics")
234- """69+def test_get_role_metrics_all(mock_get_metrics, metrics_collector):
235- Test creating MetricsCollector with default config70+ joined = (
236- """71+ "# HELP test test_metric\ntest_metric{role=\"prefill\"} 5.0\n"
237- collector = MetricsCollector() # No config passed, uses default72+ "# HELP test test_metric\ntest_metric{role=\"decode\"} 3.0"
238- 73+ )
239- # Verify default config74+ mock_get_metrics.return_value = joined
240- assert collector._cache_ttl_sec == collector.config.observability_config.metrics_ttl75+ 
241- assert isinstance(collector._lock, type(threading.RLock()))76+ result = metrics_collector.get_role_metrics()
242- assert collector._last_metrics is None77+ 
243- assert collector._last_fetch_time == 0.078+ assert "prefill" in result
79+ assert "decode" in result
80+ mock_get_metrics.assert_called_once_with(metrics_type="role", role=None)
244 81 
245 82 
246-@patch("motor.controller.observability.metrics.metrics_collector.time.monotonic")83+@patch("motor.controller.observability.metrics.metrics_collector.CoordinatorApiClient.get_metrics")
247-@patch("motor.controller.observability.metrics.metrics_collector.CoordinatorApiClient.get_full_metrics")84+def test_get_role_metrics_specific_role(mock_get_metrics, metrics_collector):
248-def test_get_full_metrics_edge_case_exact_ttl(mock_get_metrics, mock_monotonic, metrics_collector):85+ mock_get_metrics.return_value = (
249- """86+ '# HELP test test_metric\n'
250- Test edge case: Exactly reaching TTL87+ '# TYPE test_metric gauge\n'
251- """88+ 'test_metric{role="prefill"} 5.0'
252- # First call89+ )
253- mock_get_metrics.return_value = TEST_METRICS_DATA90+ 
254- mock_monotonic.return_value = 1000.091+ result = metrics_collector.get_role_metrics(role="prefill")
255- result1 = metrics_collector.get_full_metrics()92+ 
256- 93+ assert "prefill" in result
257- assert result1 == TEST_METRICS_DATA94+ mock_get_metrics.assert_called_once_with(metrics_type="role", role="prefill")
258- 95+ 
259- # Exactly at TTL (exactly 2 seconds), should refetch96+ 
260- mock_get_metrics.reset_mock()97+@patch("motor.controller.observability.metrics.metrics_collector.CoordinatorApiClient.get_metrics")
261- NEW_METRICS_DATA = TEST_METRICS_DATA + """98+def test_get_role_metrics_api_failure(mock_get_metrics, metrics_collector):
262-# Add new metric99+ mock_get_metrics.return_value = None
263-vllm:request_latency_bucket{le="0.1"} 10.0"""100+ 
264- mock_get_metrics.return_value = NEW_METRICS_DATA101+ result = metrics_collector.get_role_metrics()
265- mock_monotonic.return_value = 1002.0 # Exactly 2 seconds102+ 
266- 103+ assert result == ""
267- result2 = metrics_collector.get_full_metrics()104+ mock_get_metrics.assert_called_once_with(metrics_type="role", role=None)
268-
269- # Verify: Should refetch
270- assert result2 == NEW_METRICS_DATA
271- mock_get_metrics.assert_called_once()
Atests/controller/observability/test_observability.py+135-0
@@ -0,0 +1,135 @@
1+# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.
2+# MindIE is licensed under Mulan PSL v2.
3+# You can use this software according to the terms and conditions of the Mulan PSL v2.
4+# You may obtain a copy of Mulan PSL v2 at:
5+# http://license.coscl.org.cn/MulanPSL2
6+# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
7+# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
8+# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
9+# See the Mulan PSL v2 for more details.
10+ 
11+from unittest.mock import patch
12+import pytest
13+ 
14+from motor.common.utils.singleton import ThreadSafeSingleton
15+from motor.config.controller import ControllerConfig
16+from motor.controller.observability.observability import Observability
17+ 
18+ 
19+_FULL_METRICS = "# HELP test_metric Test metric\n# TYPE test_metric gauge\ntest_metric 1.0\n"
20+_ROLE_METRICS = {
21+ "prefill": "# HELP pref metric\n# TYPE pref gauge\npref 1.0\n",
22+ "decode": "# HELP dec metric\n# TYPE dec gauge\ndec 2.0\n",
23+}
24+ 
25+ 
26+def _cleanup_singletons():
27+ """Clean up singleton instances to ensure test isolation."""
28+ singletons_to_cleanup = [Observability]
29+ for singleton_cls in singletons_to_cleanup:
30+ if singleton_cls in ThreadSafeSingleton._instances:
31+ try:
32+ instance = ThreadSafeSingleton._instances[singleton_cls]
33+ if hasattr(instance, "stop"):
34+ instance.stop()
35+ except Exception:
36+ pass
37+ del ThreadSafeSingleton._instances[singleton_cls]
38+ 
39+ 
40+@pytest.fixture(autouse=True)
41+def cleanup_singletons():
42+ """Auto cleanup singletons before and after each test."""
43+ _cleanup_singletons()
44+ yield
45+ _cleanup_singletons()
46+ 
47+ 
48+@pytest.fixture
49+def observability():
50+ config = ControllerConfig()
51+ return Observability(config)
52+ 
53+ 
54+@patch(
55+ "motor.controller.observability.observability.CoordinatorApiClient.get_metrics"
56+)
57+def test_get_metrics_full_default(mock_get_metrics, observability):
58+ mock_get_metrics.return_value = _FULL_METRICS
59+ result = observability.get_metrics()
60+ assert result == _FULL_METRICS
61+ mock_get_metrics.assert_called_once_with(metrics_type="full", role=None)
62+ 
63+ 
64+@patch(
65+ "motor.controller.observability.observability.CoordinatorApiClient.get_metrics"
66+)
67+def test_get_metrics_full_explicit(mock_get_metrics, observability):
68+ mock_get_metrics.return_value = _FULL_METRICS
69+ result = observability.get_metrics(metrics_type="full")
70+ assert result == _FULL_METRICS
71+ mock_get_metrics.assert_called_once_with(metrics_type="full", role=None)
72+ 
73+ 
74+@patch(
75+ "motor.controller.observability.observability.CoordinatorApiClient.get_metrics"
76+)
77+def test_get_metrics_instance(mock_get_metrics, observability):
78+ mock_get_metrics.return_value = (
79+ '# HELP a\n# TYPE a gauge\na{instance_id="1",role="prefill"} 1.0\n'
80+ )
81+ result = observability.get_metrics(metrics_type="instance")
82+ assert result.startswith("# HELP")
83+ mock_get_metrics.assert_called_once_with(metrics_type="instance", role=None)
84+ 
85+ 
86+@patch(
87+ "motor.controller.observability.observability.CoordinatorApiClient.get_metrics"
88+)
89+def test_get_metrics_role_all(mock_get_metrics, observability):
90+ joined = "\n".join(_ROLE_METRICS.values())
91+ mock_get_metrics.return_value = joined
92+ result = observability.get_metrics(metrics_type="role")
93+ assert result == joined
94+ assert "# HELP pref metric" in result
95+ assert "# HELP dec metric" in result
96+ mock_get_metrics.assert_called_once_with(metrics_type="role", role=None)
97+ 
98+ 
99+@patch(
100+ "motor.controller.observability.observability.CoordinatorApiClient.get_metrics"
101+)
102+def test_get_metrics_role_filtered(mock_get_metrics, observability):
103+ mock_get_metrics.return_value = _ROLE_METRICS["prefill"]
104+ result = observability.get_metrics(metrics_type="role", role="prefill")
105+ assert result == _ROLE_METRICS["prefill"]
106+ mock_get_metrics.assert_called_once_with(metrics_type="role", role="prefill")
107+ 
108+ 
109+@patch(
110+ "motor.controller.observability.observability.CoordinatorApiClient.get_metrics"
111+)
112+def test_get_metrics_role_not_found(mock_get_metrics, observability):
113+ mock_get_metrics.return_value = ""
114+ result = observability.get_metrics(metrics_type="role", role="nonexistent")
115+ assert result == ""
116+ mock_get_metrics.assert_called_once_with(metrics_type="role", role="nonexistent")
117+ 
118+ 
119+@patch(
120+ "motor.controller.observability.observability.CoordinatorApiClient.get_metrics"
121+)
122+def test_get_metrics_api_returns_none(mock_get_metrics, observability):
123+ mock_get_metrics.return_value = None
124+ result = observability.get_metrics()
125+ assert result == ""
126+ mock_get_metrics.assert_called_once_with(metrics_type="full", role=None)
127+ 
128+ 
129+@patch(
130+ "motor.controller.observability.observability.CoordinatorApiClient.get_metrics"
131+)
132+def test_get_metrics_exception(mock_get_metrics, observability):
133+ mock_get_metrics.side_effect = RuntimeError("boom")
134+ result = observability.get_metrics()
135+ assert result == ""