Coverage for streamwise/service_manager.py: 49%
222 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-09 04:47 +0000
1"""
2Kubernetes Service Manager.
3"""
5import sys
6import re
7import logging
8import asyncio
10from http import HTTPStatus
12from aiohttp import ClientTimeout
13from asyncio import TimeoutError
15from typing import List
16from typing import Dict
17from typing import Optional
18from typing import Any
20from kubernetes_asyncio.client import ApiClient
21from kubernetes_asyncio.client import CoreV1Api
22from kubernetes_asyncio.client.exceptions import ApiException
24import http_session_manager
26sys.path.append("..")
27from k8s_utils import load_k8s_config
28from k8s_utils import parse_k8s_resource_quantity
29from k8s_utils import parse_mig_resources
30from quart_utils import get_friendly_container_name
31from quart_utils import get_class_emoji
32from streamwise_apps import VLLM_SERVICES
35async def get_k8s_pod_events(
36 k8s_api: CoreV1Api,
37 namespace: str,
38 pod_name: str
39) -> List[Dict]:
40 """Fetch events related to a specific pod in a namespace."""
41 pod_field_selector = f"involvedObject.name={pod_name},involvedObject.namespace={namespace}"
42 try:
43 events = await k8s_api.list_namespaced_event(
44 namespace=namespace,
45 field_selector=pod_field_selector
46 )
47 return [{
48 "last_timestamp": event.last_timestamp,
49 "reason": event.reason,
50 "message": event.message,
51 "type": event.type,
52 "count": event.count,
53 } for event in events.items]
54 except ApiException as ex:
55 logging.error(f"Cannot read events for pod {pod_name}: {ex.reason}.")
56 except Exception as ex:
57 logging.error(f"Cannot read events for pod {pod_name}: {ex}")
58 return []
61async def get_service_files(
62 container_name: str,
63 url: Optional[str] = None
64) -> Optional[List[str]]:
65 """Asynchronous version of get_service_files"""
66 if url is None or url == "N/A":
67 return None
68 if container_name in VLLM_SERVICES:
69 return []
71 timeout = ClientTimeout(total=0.5, connect=0.5) # Short timeout for faster responses
72 try:
73 session = await http_session_manager.get_global_session()
74 async with session.get(f"{url}/files", timeout=timeout) as response:
75 content_type = response.headers.get("Content-Type", "")
76 if response.status == HTTPStatus.OK:
77 if "application/json" in content_type:
78 content_json = await response.json()
79 return content_json.get("files", [])
80 logging.warning(f"Unexpected response from '{container_name}' on '{url}/files': "
81 f"status={response.status}, content-type={content_type}")
82 except TimeoutError:
83 logging.warning(f"Timeout fetching files for {url}.")
84 except Exception as ex:
85 logging.error(f"Error fetching files for {url}: {ex}.")
86 return []
89def parse_vllm_metrics(metrics_text: str) -> Dict[str, float]:
90 """
91 Parses vLLM metrics from the provided text.
92 https://docs.vllm.ai/en/stable/design/v1/metrics.html#v0-metrics
93 Example metric lines:
94 vllm:num_requests_running{engine="0",model_name="google/gemma-3-27b-it"} 0.0
95 vllm:request_success_total{engine="0",finished_reason="stop",model_name="google/gemma-3-27b-it"} 0.0
96 vllm:request_prompt_tokens_bucket{engine="0",le="1.0",model_name="google/gemma-3-27b-it"} 0.0
97 vllm:request_success_total{engine="0",finished_reason="stop",model_name="google/gemma-3-27b-it"} 0.0
98 vllm:request_success_total{engine="0",finished_reason="length",model_name="google/gemma-3-27b-it"} 0.0
99 vllm:request_success_total{engine="0",finished_reason="abort",model_name="google/gemma-3-27b-it"} 0.0
100 """
101 metrics: Dict[str, float] = {}
102 for line in metrics_text.splitlines():
103 if not line or line.startswith("#"):
104 continue
105 m = re.match(r"^vllm:([a-zA-Z_][a-zA-Z0-9_:]*)(\{.*?\})?\s+([0-9.]+)$", line)
106 if m:
107 metric_name = m.group(1)
108 # labels = m.group(2)
109 value = float(m.group(3))
110 metrics[metric_name] = metrics.get(metric_name, 0) + value
111 return metrics
114async def get_service_health(
115 container_name: str,
116 url: str
117) -> Optional[Dict[str, Any]]:
118 """Get the health status of a service asynchronously."""
119 if url is None or url == "N/A":
120 return None
121 try:
122 timeout = ClientTimeout(total=0.5, connect=0.5) # Short timeout for faster responses
123 session = await http_session_manager.get_global_session()
124 # vLLM services (Gemma, Llama, Whisper) use /metrics instead of /health
125 if container_name in VLLM_SERVICES:
126 # /load
127 # /v1/models
128 # /version
129 # /metrics
130 async with session.get(f"{url}/metrics", timeout=timeout) as response:
131 if response.status == HTTPStatus.OK:
132 text = await response.text()
133 vllm_metrics = parse_vllm_metrics(text)
134 is_running = vllm_metrics.get("num_requests_running", 0) > 0
135 return {
136 "status": "ok",
137 "running": is_running, # It doesn't block
138 "vllm_metrics": vllm_metrics,
139 }
140 return {"status": f"unhealthy ({response.status})"}
141 # Rest of the services
142 else:
143 async with session.get(f"{url}/health", timeout=timeout) as response:
144 if response.status == HTTPStatus.OK:
145 content_json = await response.json()
146 if len(content_json) == 1 and "health" not in content_json:
147 # This is a nested health response, e.g. {"service_name": { ... }}
148 for _, value in content_json.items():
149 return value
150 return content_json
151 return {"status": f"Unhealthy ({response.status})"}
152 except TimeoutError:
153 logging.warning(f"Timeout checking health for {url}.")
154 return {"status": "timeout"}
155 except Exception as ex:
156 logging.error(f"Error checking health for {container_name} on {url}: {ex}.")
157 return {"status": "failed"}
160async def get_health_and_files_async(
161 services_data: List[Dict[str, Any]]
162) -> List[Dict[str, Any]]:
163 """Get health and files for multiple services asynchronously."""
164 if not services_data:
165 return services_data
166 for service in services_data:
167 service["health"] = "N/A"
168 service["files"] = []
170 async def fetch_for_service(service: Dict[str, Any]) -> Dict[str, Any]:
171 url = service.get("url", "N/A")
172 if not url or url == "N/A":
173 return service
174 container_name = service.get("container_name", "unknown")
175 try:
176 health_task = get_service_health(container_name, url)
177 files_task = get_service_files(container_name, url)
178 health, files = await asyncio.gather(
179 health_task,
180 files_task,
181 return_exceptions=True)
183 if isinstance(health, Exception):
184 logging.warning(f"Health check failed for {container_name} at {url}: {health}")
185 service["health"] = health
187 if isinstance(files, Exception):
188 logging.warning(f"Fetching files failed for {container_name} at {url}: {files}")
189 else:
190 service["files"] = files
191 except TimeoutError:
192 logging.warning(f"Timeout fetching data for {container_name} at {url}.")
193 service["health"] = "Timeout"
194 except Exception as ex:
195 logging.error(f"Error fetching data for {container_name} at {url}: {ex}.")
196 service["health"] = "Error"
197 return service
199 return await asyncio.gather(*(fetch_for_service(svc) for svc in services_data))
202async def get_k8s_container_logs(
203 k8s_api: CoreV1Api,
204 namespace: str,
205 pod_name: str,
206 container_name: str,
207 num_lines: int = 500,
208 follow: bool = False
209) -> Optional[str]:
210 """Fetch logs for a specific container in a pod within a namespace."""
211 try:
212 return await k8s_api.read_namespaced_pod_log(
213 name=pod_name,
214 namespace=namespace,
215 container=container_name,
216 follow=follow,
217 tail_lines=num_lines,
218 _request_timeout=1.0, # Short timeout for faster UI responses
219 )
220 except ApiException as ex:
221 logging.error(f"Cannot read logs for pod {pod_name}/{container_name}: {ex.reason}.")
222 except Exception as ex:
223 logging.error(f"Cannot read logs for pod {pod_name}/{container_name} [{type(ex)}]: {ex}.")
224 return None
227async def get_services_ns(
228 namespace: str = "default",
229 container_name_filter: Optional[str] = None,
230 details: bool = False,
231 k8s_cluster: Optional[str] = None
232) -> List[Dict]:
233 """Get all services in a specific namespace."""
234 ret = []
236 await load_k8s_config(k8s_cluster)
237 async with ApiClient() as api_client:
238 k8s_api = CoreV1Api(api_client)
239 pods = await k8s_api.list_namespaced_pod(namespace)
240 for pod in pods.items:
241 pod_name = pod.metadata.name
242 namespace = pod.metadata.namespace
243 pod_status = pod.status.phase
244 start_time = pod.status.start_time
246 if pod.status.container_statuses is not None:
247 for container_status in pod.status.container_statuses:
248 container_state = container_status.state
249 if container_state.running is not None:
250 # container_state.started_at
251 pod_status = "Running"
252 if container_state.terminated is not None:
253 # container_state.terminated
254 pod_status = "Terminated"
255 if container_state.waiting is not None:
256 pod_status = container_state.waiting.reason
258 # Events
259 events_list = await get_k8s_pod_events(
260 k8s_api,
261 namespace,
262 pod_name) if details else []
264 pod_ip = pod.status.pod_ip
265 node_name = pod.spec.node_name
266 for container in pod.spec.containers:
267 container_name = container.name
268 if container_name_filter and container_name != container_name_filter:
269 continue
271 image = container.image
273 # Resources
274 resources = container.resources.requests
275 cpu: int | float = 0
276 memory: int | float = 0
277 ephemeral_storage: int | float = 0
278 gpu: int = 0
279 mig_profile: Optional[str] = None
280 if resources is not None:
281 cpu = parse_k8s_resource_quantity(resources.get("cpu", "0"))
282 memory = parse_k8s_resource_quantity(resources.get("memory", "0"))
283 ephemeral_storage = parse_k8s_resource_quantity(resources.get("ephemeral-storage", "0"))
284 gpu = int(resources.get("nvidia.com/gpu", 0))
286 mig_resources = parse_mig_resources(resources)
287 if mig_resources:
288 mig_profile = next(iter(mig_resources))
289 gpu = mig_resources[mig_profile]
291 # Logs
292 logs = await get_k8s_container_logs(
293 k8s_api,
294 namespace,
295 pod_name,
296 container_name
297 ) if details else None
299 if pod_ip is None or not container.ports:
300 ret.append({
301 "namespace": namespace,
302 "pod_name": pod_name,
303 "pod_ip": pod_ip,
304 "container_port": None,
305 "container_name": container_name,
306 "pod_status": pod_status,
307 "start_time": start_time,
308 "url": "N/A",
309 "node_name": node_name,
310 "cpu": cpu,
311 "memory": memory,
312 "gpu": gpu,
313 "mig_profile": mig_profile,
314 "ephemeral_storage": ephemeral_storage,
315 "events": events_list,
316 "image": image,
317 "logs": logs,
318 "health": None,
319 "files": None,
320 })
321 else:
322 for container_port in container.ports:
323 url = f"{http_session_manager.SERVICE_SCHEME}://{pod_ip}:{container_port.container_port}"
325 ret.append({
326 "namespace": namespace,
327 "pod_name": pod_name,
328 "pod_ip": pod_ip,
329 "container_port": container_port.container_port,
330 "container_name": container_name,
331 "pod_status": pod_status,
332 "start_time": start_time,
333 "url": url,
334 "node_name": node_name,
335 "cpu": cpu,
336 "memory": memory,
337 "gpu": gpu,
338 "mig_profile": mig_profile,
339 "ephemeral_storage": ephemeral_storage,
340 "events": events_list,
341 "image": image,
342 "logs": logs,
343 "health": None, # Populated asynchronously
344 "files": None, # Populated asynchronously
345 })
347 # Run async health and files fetching if there are services with URLs
348 services_with_urls = [
349 service
350 for service in ret
351 if service["url"] != "N/A"
352 ]
353 if services_with_urls:
354 await get_health_and_files_async(services_with_urls)
356 return ret
359async def get_services(
360 container_name_filter: Optional[str] = None,
361 details: bool = False,
362 namespace: str = "default",
363 k8s_cluster: Optional[str] = None
364) -> List[Dict]:
365 """Get all services across all namespaces."""
366 ret = []
368 await load_k8s_config(k8s_cluster)
369 async with ApiClient() as api_client:
370 k8s_api = CoreV1Api(api_client)
371 try:
372 namespace_list = await k8s_api.list_namespace()
373 for ns in namespace_list.items:
374 namespace_ix = ns.metadata.name
375 if namespace_ix.startswith(namespace):
376 services = await get_services_ns(
377 namespace,
378 container_name_filter=container_name_filter,
379 details=details,
380 k8s_cluster=k8s_cluster)
381 if services:
382 ret.extend(services)
383 except ApiException as api_ex:
384 print(f"Exception when listing namespace '{namespace}': {api_ex}")
385 return ret
388async def get_service_timestamps(
389 pod_name: str,
390 container_name: str,
391 url: str
392) -> Optional[List[Dict]]:
393 """Get the timestamps from a service asynchronously."""
394 if url is None or url == "N/A":
395 return None
396 if container_name in VLLM_SERVICES:
397 return []
399 try:
400 session = await http_session_manager.get_global_session()
401 timeout = ClientTimeout(total=0.5, connect=0.5) # Short timeout for faster responses
402 async with session.get(f"{url}/timestamps", timeout=timeout) as response:
403 if response.status == HTTPStatus.OK:
404 content_json = await response.json()
405 if len(content_json) == 1:
406 for _, timestamps in content_json.items():
407 for timestamp in timestamps:
408 original_id = timestamp.get("id", "")
409 timestamp["id"] = f"{pod_name}_{original_id}"
410 service_name = timestamp["group"]
411 container_name = await get_friendly_container_name(service_name)
412 class_emoji = await get_class_emoji(service_name)
413 timestamp["group"] = f"{container_name} {class_emoji}"
414 return timestamps
415 return content_json
416 except Exception as ex:
417 logging.error(f"Error checking timestamps for {url}: {ex}.")
418 return []