wehub-resource-sync
148 lines
5.7 KiB
Python
148 lines
5.7 KiB
Python
import asyncio
|
|
import sys
|
|
import time
|
|
from collections import deque
|
|
from dataclasses import dataclass, field
|
|
from typing import Deque
|
|
|
|
from ray.serve._private.constants import CONTROL_LOOP_INTERVAL_S
|
|
from ray.serve.schema import ControllerHealthMetrics, DurationStats
|
|
|
|
# Number of recent loop iterations to track for rolling averages
|
|
_HEALTH_METRICS_HISTORY_SIZE = 100
|
|
|
|
|
|
@dataclass
|
|
class ControllerHealthMetricsTracker:
|
|
"""Tracker for collecting controller health metrics over time."""
|
|
|
|
controller_start_time: float = field(default_factory=time.time)
|
|
|
|
# Rolling history of loop durations
|
|
loop_durations: Deque[float] = field(
|
|
default_factory=lambda: deque(maxlen=_HEALTH_METRICS_HISTORY_SIZE)
|
|
)
|
|
|
|
# Rolling history of metrics delays
|
|
handle_metrics_delays: Deque[float] = field(
|
|
default_factory=lambda: deque(maxlen=_HEALTH_METRICS_HISTORY_SIZE)
|
|
)
|
|
replica_metrics_delays: Deque[float] = field(
|
|
default_factory=lambda: deque(maxlen=_HEALTH_METRICS_HISTORY_SIZE)
|
|
)
|
|
|
|
# Rolling history of component update durations
|
|
dsm_update_durations: Deque[float] = field(
|
|
default_factory=lambda: deque(maxlen=_HEALTH_METRICS_HISTORY_SIZE)
|
|
)
|
|
asm_update_durations: Deque[float] = field(
|
|
default_factory=lambda: deque(maxlen=_HEALTH_METRICS_HISTORY_SIZE)
|
|
)
|
|
proxy_update_durations: Deque[float] = field(
|
|
default_factory=lambda: deque(maxlen=_HEALTH_METRICS_HISTORY_SIZE)
|
|
)
|
|
node_update_durations: Deque[float] = field(
|
|
default_factory=lambda: deque(maxlen=_HEALTH_METRICS_HISTORY_SIZE)
|
|
)
|
|
|
|
# Latest values (used in collect_metrics)
|
|
last_sleep_duration_s: float = 0.0
|
|
num_control_loops: int = 0
|
|
last_control_loop_time: float = 0.0
|
|
|
|
def record_loop_duration(self, duration: float):
|
|
self.loop_durations.append(duration)
|
|
|
|
def record_handle_metrics_delay(self, delay_ms: float):
|
|
self.handle_metrics_delays.append(delay_ms)
|
|
|
|
def record_replica_metrics_delay(self, delay_ms: float):
|
|
self.replica_metrics_delays.append(delay_ms)
|
|
|
|
def record_dsm_update_duration(self, duration: float):
|
|
self.dsm_update_durations.append(duration)
|
|
|
|
def record_asm_update_duration(self, duration: float):
|
|
self.asm_update_durations.append(duration)
|
|
|
|
def record_proxy_update_duration(self, duration: float):
|
|
self.proxy_update_durations.append(duration)
|
|
|
|
def record_node_update_duration(self, duration: float):
|
|
self.node_update_durations.append(duration)
|
|
|
|
def collect_metrics(self) -> ControllerHealthMetrics:
|
|
"""Collect and return current health metrics."""
|
|
now = time.time()
|
|
|
|
# Calculate loop statistics from rolling history
|
|
loop_duration_stats = DurationStats.from_values(list(self.loop_durations))
|
|
|
|
# Calculate loops per second based on uptime and total loops
|
|
uptime = now - self.controller_start_time
|
|
loops_per_second = self.num_control_loops / uptime if uptime > 0 else 0.0
|
|
|
|
# Calculate event loop delay (actual sleep - expected sleep)
|
|
# Positive values indicate the event loop is overloaded
|
|
event_loop_delay = max(
|
|
0.0, self.last_sleep_duration_s - CONTROL_LOOP_INTERVAL_S
|
|
)
|
|
|
|
# Get asyncio task count
|
|
try:
|
|
loop = asyncio.get_event_loop()
|
|
num_asyncio_tasks = len(asyncio.all_tasks(loop))
|
|
except RuntimeError:
|
|
num_asyncio_tasks = 0
|
|
|
|
# Calculate metrics delay statistics
|
|
handle_delay_stats = DurationStats.from_values(list(self.handle_metrics_delays))
|
|
replica_delay_stats = DurationStats.from_values(
|
|
list(self.replica_metrics_delays)
|
|
)
|
|
|
|
# Calculate component update duration statistics
|
|
dsm_update_stats = DurationStats.from_values(list(self.dsm_update_durations))
|
|
asm_update_stats = DurationStats.from_values(list(self.asm_update_durations))
|
|
proxy_update_stats = DurationStats.from_values(
|
|
list(self.proxy_update_durations)
|
|
)
|
|
node_update_stats = DurationStats.from_values(list(self.node_update_durations))
|
|
|
|
# Get memory usage in MB
|
|
# Note: ru_maxrss is in bytes on macOS but kilobytes on Linux
|
|
# The resource module is Unix-only, so we handle Windows gracefully
|
|
try:
|
|
import resource
|
|
|
|
rusage = resource.getrusage(resource.RUSAGE_SELF)
|
|
process_memory_mb = (
|
|
rusage.ru_maxrss / (1024 * 1024) # Convert bytes to MB on macOS
|
|
if sys.platform == "darwin"
|
|
else rusage.ru_maxrss / 1024 # Convert KB to MB on Linux
|
|
)
|
|
except ImportError:
|
|
# resource module not available on Windows
|
|
process_memory_mb = 0.0
|
|
|
|
return ControllerHealthMetrics(
|
|
timestamp=now,
|
|
controller_start_time=self.controller_start_time,
|
|
uptime_s=uptime,
|
|
last_control_loop_time=self.last_control_loop_time,
|
|
num_control_loops=self.num_control_loops,
|
|
loop_duration_s=loop_duration_stats,
|
|
loops_per_second=loops_per_second,
|
|
last_sleep_duration_s=self.last_sleep_duration_s,
|
|
expected_sleep_duration_s=CONTROL_LOOP_INTERVAL_S,
|
|
event_loop_delay_s=event_loop_delay,
|
|
num_asyncio_tasks=num_asyncio_tasks,
|
|
deployment_state_update_duration_s=dsm_update_stats,
|
|
application_state_update_duration_s=asm_update_stats,
|
|
proxy_state_update_duration_s=proxy_update_stats,
|
|
node_update_duration_s=node_update_stats,
|
|
handle_metrics_delay_ms=handle_delay_stats,
|
|
replica_metrics_delay_ms=replica_delay_stats,
|
|
process_memory_mb=process_memory_mb,
|
|
)
|