diff --git a/backend/package/yuxi/channel/monitoring/__init__.py b/backend/package/yuxi/channel/monitoring/__init__.py new file mode 100644 index 00000000..9efc002e --- /dev/null +++ b/backend/package/yuxi/channel/monitoring/__init__.py @@ -0,0 +1,123 @@ +from yuxi.channel.monitoring.diagnostic_heartbeat import ( + DiagnosticHeartbeat, + HeartbeatReport, + MemoryPressureEvent, + MemorySnapshot, + diagnostic_heartbeat, +) +from yuxi.channel.monitoring.diagnostic_phase import ( + PhaseSnapshot, + diagnostic_phase, + get_current_phase, + get_recent_phases, + record_diagnostic_phase, + reset_diagnostic_phases_for_test, +) +from yuxi.channel.monitoring.diagnostic_stability import ( + StabilityEvent, + StabilityRingBuffer, + get_stability_snapshot, + install_stability_crash_hook, + record_stability_event, + write_stability_bundle, +) +from yuxi.channel.monitoring.event_loop_monitor import ( + EventLoopHealth, + EventLoopMonitor, + EventLoopStats, + event_loop_monitor, +) +from yuxi.channel.monitoring.event_loop_ready import ( + EventLoopReadyResult, + wait_for_event_loop_ready, +) +from yuxi.channel.monitoring.health_monitor import ( + ChannelHealthEvaluation, + ChannelHealthMetrics, + ChannelHealthMonitor, + channel_health_monitor, + evaluate_channel_health, + resolve_restart_reason, +) +from yuxi.channel.monitoring.metrics import ( + metrics_registry, + send_alert_webhook, + active_sessions, + active_tasks, + channel_health, + event_loop_delay_p99, + event_loop_utilization, + memory_gc_objects, + memory_rss_mb, + stuck_sessions, +) +from yuxi.channel.monitoring.readiness import ( + ReadinessChecker, + ReadinessResult, + readiness_checker, +) +from yuxi.channel.monitoring.state_aggregator import ( + ChannelHealthState, + ConfigHealthState, + DiagnosticState, + GatewayConnectionState, + MonitoringState, + ReadinessState, + StateAggregator, + UnifiedState, + state_aggregator, +) + +__all__ = [ + "ChannelHealthEvaluation", + "ChannelHealthMetrics", + "ChannelHealthMonitor", + "ChannelHealthState", + "ConfigHealthState", + "DiagnosticHeartbeat", + "DiagnosticState", + "EventLoopHealth", + "EventLoopMonitor", + "EventLoopReadyResult", + "EventLoopStats", + "GatewayConnectionState", + "HeartbeatReport", + "MemoryPressureEvent", + "MemorySnapshot", + "MonitoringState", + "PhaseSnapshot", + "ReadinessChecker", + "ReadinessResult", + "ReadinessState", + "StabilityEvent", + "StabilityRingBuffer", + "StateAggregator", + "UnifiedState", + "channel_health_monitor", + "diagnostic_heartbeat", + "diagnostic_phase", + "event_loop_monitor", + "evaluate_channel_health", + "get_current_phase", + "get_recent_phases", + "get_stability_snapshot", + "install_stability_crash_hook", + "readiness_checker", + "record_diagnostic_phase", + "record_stability_event", + "reset_diagnostic_phases_for_test", + "resolve_restart_reason", + "state_aggregator", + "wait_for_event_loop_ready", + "write_stability_bundle", + "metrics_registry", + "send_alert_webhook", + "active_sessions", + "active_tasks", + "channel_health", + "event_loop_delay_p99", + "event_loop_utilization", + "memory_gc_objects", + "memory_rss_mb", + "stuck_sessions", +] diff --git a/backend/package/yuxi/channel/monitoring/diagnostic_heartbeat.py b/backend/package/yuxi/channel/monitoring/diagnostic_heartbeat.py new file mode 100644 index 00000000..9d602a80 --- /dev/null +++ b/backend/package/yuxi/channel/monitoring/diagnostic_heartbeat.py @@ -0,0 +1,473 @@ +import asyncio +import gc +import logging +import time +from dataclasses import dataclass, field + +logger = logging.getLogger(__name__) + +DEFAULT_HEARTBEAT_INTERVAL = 30 +STUCK_SESSION_TIMEOUT = 600 +MAX_CONSECUTIVE_FAILURES = 5 + +MB = 1024 * 1024 +DEFAULT_RSS_WARNING_MB = 1536 +DEFAULT_RSS_CRITICAL_MB = 3072 +DEFAULT_DATA_WARNING_MB = 1024 +DEFAULT_DATA_CRITICAL_MB = 2048 +DEFAULT_GC_OBJECTS_WARNING = 500000 +DEFAULT_GC_SAMPLE_INTERVAL = 5 +DEFAULT_RSS_GROWTH_WARNING_MB = 512 +DEFAULT_RSS_GROWTH_CRITICAL_MB = 1024 +DEFAULT_GROWTH_WINDOW_S = 600 +DEFAULT_PRESSURE_COOLDOWN_S = 300 + + +@dataclass +class MemorySnapshot: + rss_mb: float = 0.0 + vms_mb: float = 0.0 + data_mb: float = 0.0 + gc_objects: int = 0 + gc_g0_count: int = 0 + gc_g1_count: int = 0 + gc_g2_count: int = 0 + timestamp: float = 0.0 + + +@dataclass +class MemoryPressureEvent: + level: str # "warning" | "critical" + reason: str # "rss_threshold" | "data_threshold" | "rss_growth" + rss_mb: float = 0.0 + threshold_mb: float = 0.0 + rss_growth_mb: float = 0.0 + window_s: float = 0.0 + timestamp: float = 0.0 + + +@dataclass +class HeartbeatReport: + active_sessions: int = 0 + stuck_sessions: int = 0 + stuck_details: list[dict] = field(default_factory=list) + tool_loop_sessions: int = 0 + tool_loop_details: list[dict] = field(default_factory=list) + memory: MemorySnapshot = field(default_factory=MemorySnapshot) + active_tasks: int = 0 + uptime_seconds: float = 0.0 + process_start_timestamp: float = 0.0 + timestamp: float = 0.0 + warnings: list[str] = field(default_factory=list) + memory_pressure: MemoryPressureEvent | None = None + + def to_dict(self) -> dict: + result = { + "active_sessions": self.active_sessions, + "stuck_sessions": self.stuck_sessions, + "stuck_details": self.stuck_details, + "tool_loop_sessions": self.tool_loop_sessions, + "tool_loop_details": self.tool_loop_details, + "memory": { + "rss_mb": self.memory.rss_mb, + "vms_mb": self.memory.vms_mb, + "data_mb": self.memory.data_mb, + "gc_objects": self.memory.gc_objects, + "gc_g0_count": self.memory.gc_g0_count, + "gc_g1_count": self.memory.gc_g1_count, + "gc_g2_count": self.memory.gc_g2_count, + "timestamp": self.memory.timestamp, + }, + "active_tasks": self.active_tasks, + "uptime_seconds": self.uptime_seconds, + "process_start_timestamp": self.process_start_timestamp, + "timestamp": self.timestamp, + "warnings": self.warnings, + } + if self.memory_pressure: + result["memory_pressure"] = { + "level": self.memory_pressure.level, + "reason": self.memory_pressure.reason, + "rss_mb": self.memory_pressure.rss_mb, + "threshold_mb": self.memory_pressure.threshold_mb, + "rss_growth_mb": self.memory_pressure.rss_growth_mb, + "window_s": self.memory_pressure.window_s, + "timestamp": self.memory_pressure.timestamp, + } + return result + + +class DiagnosticHeartbeat: + """诊断心跳：30s 间隔检测卡住 session、内存快照 + 内存压力。 + + 集成到 FastAPI lifespan 后启动，可被 health endpoint 消费。 + """ + + def __init__( + self, + interval: int = DEFAULT_HEARTBEAT_INTERVAL, + stuck_timeout: int = STUCK_SESSION_TIMEOUT, + rss_warning_mb: float = DEFAULT_RSS_WARNING_MB, + rss_critical_mb: float = DEFAULT_RSS_CRITICAL_MB, + data_warning_mb: float = DEFAULT_DATA_WARNING_MB, + data_critical_mb: float = DEFAULT_DATA_CRITICAL_MB, + rss_growth_warning_mb: float = DEFAULT_RSS_GROWTH_WARNING_MB, + rss_growth_critical_mb: float = DEFAULT_RSS_GROWTH_CRITICAL_MB, + growth_window_s: float = DEFAULT_GROWTH_WINDOW_S, + pressure_cooldown_s: float = DEFAULT_PRESSURE_COOLDOWN_S, + gc_objects_warning: int = DEFAULT_GC_OBJECTS_WARNING, + gc_sample_interval: int = DEFAULT_GC_SAMPLE_INTERVAL, + ): + self._interval = interval + self._stuck_timeout = stuck_timeout + self._rss_warning_mb = rss_warning_mb + self._rss_critical_mb = rss_critical_mb + self._data_warning_mb = data_warning_mb + self._data_critical_mb = data_critical_mb + self._rss_growth_warning_mb = rss_growth_warning_mb + self._rss_growth_critical_mb = rss_growth_critical_mb + self._growth_window_s = growth_window_s + self._pressure_cooldown_s = pressure_cooldown_s + self._gc_objects_warning = gc_objects_warning + self._gc_sample_interval = gc_sample_interval + self._running = False + self._task: asyncio.Task | None = None + self._last_report: HeartbeatReport | None = None + self._last_memory: MemorySnapshot | None = None + self._last_pressure_at: dict[str, float] = {} + self._start_time: float | None = None + self._start_monotonic: float | None = None + self._consecutive_failures = 0 + self._cycle_count = 0 + + @property + def last_report(self) -> HeartbeatReport | None: + return self._last_report + + @staticmethod + def _resolve_start_time() -> float: + try: + import psutil + + return psutil.Process().create_time() + except Exception: + return time.time() + + async def start(self): + self._running = True + self._start_time = self._resolve_start_time() + self._start_monotonic = time.monotonic() + self._task = asyncio.create_task(self._loop()) + logger.info( + "DiagnosticHeartbeat started (interval=%ds, stuck_timeout=%ds)", + self._interval, + self._stuck_timeout, + ) + + async def stop(self): + self._running = False + if self._task: + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + logger.info("DiagnosticHeartbeat stopped") + + async def _loop(self): + while self._running: + await asyncio.sleep(self._interval) + try: + self._last_report = await self._collect() + self._consecutive_failures = 0 + await self._push_metrics(self._last_report) + if self._last_report.warnings: + for w in self._last_report.warnings: + logger.warning("DiagnosticHeartbeat: %s", w) + if self._last_report.memory_pressure: + mp = self._last_report.memory_pressure + logger.warning( + "DiagnosticHeartbeat: 内存压力 %s (reason=%s, rss=%.0fMB, threshold=%.0fMB)", + mp.level, + mp.reason, + mp.rss_mb, + mp.threshold_mb, + ) + except Exception: + self._consecutive_failures += 1 + logger.exception( + "DiagnosticHeartbeat: collect failed (consecutive=%d/%d)", + self._consecutive_failures, + MAX_CONSECUTIVE_FAILURES, + ) + if self._consecutive_failures >= MAX_CONSECUTIVE_FAILURES: + logger.critical( + "DiagnosticHeartbeat: %d consecutive collect failures, heartbeat may be broken", + self._consecutive_failures, + ) + + async def _collect(self) -> HeartbeatReport: + now = time.monotonic() + self._cycle_count += 1 + memory = await self._capture_memory(now) + self._last_memory = memory + + snapshots = _get_all_snapshots() + stuck_details, stuck_count = await self._detect_stuck_sessions(now, snapshots) + active_tasks = len(asyncio.all_tasks()) + active_sessions = _count_active_sessions_from(snapshots) + + uptime_seconds = now - self._start_monotonic if self._start_monotonic else 0.0 + process_start_timestamp = self._start_time or 0.0 + + memory_pressure = self._detect_memory_pressure(memory, now) + + warnings: list[str] = [] + if stuck_count > 0: + warnings.append(f"检测到 {stuck_count} 个卡住 session") + if memory.rss_mb > self._rss_warning_mb: + warnings.append(f"内存 RSS {memory.rss_mb:.0f}MB > {self._rss_warning_mb:.0f}MB") + if memory.data_mb > self._data_warning_mb: + warnings.append(f"内存 DATA {memory.data_mb:.0f}MB > {self._data_warning_mb:.0f}MB") + if memory.gc_objects > self._gc_objects_warning: + warnings.append(f"GC 对象数 {memory.gc_objects} > {self._gc_objects_warning}") + if memory_pressure: + level_label = "严重" if memory_pressure.level == "critical" else "" + warnings.append(f"内存压力{level_label}: {memory_pressure.reason} (RSS={memory_pressure.rss_mb:.0f}MB)") + + return HeartbeatReport( + active_sessions=active_sessions, + stuck_sessions=stuck_count, + stuck_details=stuck_details, + tool_loop_sessions=0, + tool_loop_details=[], + memory=memory, + active_tasks=active_tasks, + uptime_seconds=uptime_seconds, + process_start_timestamp=process_start_timestamp, + timestamp=now, + warnings=warnings, + memory_pressure=memory_pressure, + ) + + async def _push_metrics(self, report: HeartbeatReport): + try: + from yuxi.channel.monitoring.metrics import ( + active_sessions, + active_tasks, + memory_gc_objects, + memory_rss_mb, + stuck_sessions, + ) + + active_sessions.set(report.active_sessions) + stuck_sessions.set(report.stuck_sessions) + memory_rss_mb.set(report.memory.rss_mb) + memory_gc_objects.set(report.memory.gc_objects) + active_tasks.set(report.active_tasks) + except Exception: + pass + + def _detect_memory_pressure(self, current: MemorySnapshot, now: float) -> MemoryPressureEvent | None: + pressure = _check_threshold_pressure( + current, + rss_warning_mb=self._rss_warning_mb, + rss_critical_mb=self._rss_critical_mb, + ) + if pressure is None: + pressure = _check_data_threshold_pressure( + current, + data_warning_mb=self._data_warning_mb, + data_critical_mb=self._data_critical_mb, + ) + if pressure is None: + pressure = _check_growth_pressure( + self._last_memory, + current, + now, + growth_window_s=self._growth_window_s, + rss_growth_warning_mb=self._rss_growth_warning_mb, + rss_growth_critical_mb=self._rss_growth_critical_mb, + ) + + if pressure is None: + return None + + key = pressure.reason + last_at = self._last_pressure_at.get(key, 0) + if now - last_at < self._pressure_cooldown_s: + return None + + self._last_pressure_at[key] = now + return pressure + + async def _detect_stuck_sessions(self, now: float, snapshots: dict) -> tuple[list[dict], int]: + stuck: list[dict] = [] + for key, snap in snapshots.items(): + if snap.last_event_at and (now - snap.last_event_at) > self._stuck_timeout: + state_value = _get_state_value(snap.state) + if snap.connected and state_value in ("running", "retrying"): + stuck.append( + { + "key": key, + "state": state_value, + "last_event_at": snap.last_event_at, + "idle_seconds": round(now - snap.last_event_at, 1), + } + ) + return stuck, len(stuck) + + async def _capture_memory(self, now: float) -> MemorySnapshot: + snapshot = MemorySnapshot(timestamp=now) + try: + import psutil + + proc = psutil.Process() + mem = proc.memory_info() + snapshot.rss_mb = round(mem.rss / 1024 / 1024, 2) + snapshot.vms_mb = round(mem.vms / 1024 / 1024, 2) + snapshot.data_mb = round(getattr(mem, "data", 0) / 1024 / 1024, 2) + except Exception: + pass + + gc_counts = gc.get_count() + snapshot.gc_g0_count = gc_counts[0] + snapshot.gc_g1_count = gc_counts[1] + snapshot.gc_g2_count = gc_counts[2] + + if self._cycle_count % self._gc_sample_interval == 0: + snapshot.gc_objects = len(gc.get_objects()) + + return snapshot + + def get_memory_trend(self) -> dict: + if self._last_memory: + return { + "rss_mb": self._last_memory.rss_mb, + "vms_mb": self._last_memory.vms_mb, + "data_mb": self._last_memory.data_mb, + "gc_objects": self._last_memory.gc_objects, + "timestamp": self._last_memory.timestamp, + } + return {} + + def reset(self): + self._last_memory = None + self._last_pressure_at.clear() + self._last_report = None + self._cycle_count = 0 + + +def _check_threshold_pressure( + memory: MemorySnapshot, + *, + rss_warning_mb: float, + rss_critical_mb: float, +) -> MemoryPressureEvent | None: + if memory.rss_mb >= rss_critical_mb: + return MemoryPressureEvent( + level="critical", + reason="rss_threshold", + rss_mb=memory.rss_mb, + threshold_mb=rss_critical_mb, + timestamp=memory.timestamp, + ) + if memory.rss_mb >= rss_warning_mb: + return MemoryPressureEvent( + level="warning", + reason="rss_threshold", + rss_mb=memory.rss_mb, + threshold_mb=rss_warning_mb, + timestamp=memory.timestamp, + ) + return None + + +def _check_data_threshold_pressure( + memory: MemorySnapshot, + *, + data_warning_mb: float, + data_critical_mb: float, +) -> MemoryPressureEvent | None: + if memory.data_mb >= data_critical_mb: + return MemoryPressureEvent( + level="critical", + reason="data_threshold", + rss_mb=memory.rss_mb, + threshold_mb=data_critical_mb, + timestamp=memory.timestamp, + ) + if memory.data_mb >= data_warning_mb: + return MemoryPressureEvent( + level="warning", + reason="data_threshold", + rss_mb=memory.rss_mb, + threshold_mb=data_warning_mb, + timestamp=memory.timestamp, + ) + return None + + +def _check_growth_pressure( + previous: MemorySnapshot | None, + current: MemorySnapshot, + now: float, + *, + growth_window_s: float, + rss_growth_warning_mb: float, + rss_growth_critical_mb: float, +) -> MemoryPressureEvent | None: + if previous is None or previous.rss_mb <= 0: + return None + + window_s = now - previous.timestamp + if window_s <= 0 or window_s > growth_window_s: + return None + + rss_growth_mb = current.rss_mb - previous.rss_mb + if rss_growth_mb <= 0: + return None + + if rss_growth_mb >= rss_growth_critical_mb: + return MemoryPressureEvent( + level="critical", + reason="rss_growth", + rss_mb=current.rss_mb, + threshold_mb=rss_growth_critical_mb, + rss_growth_mb=round(rss_growth_mb, 2), + window_s=round(window_s, 1), + timestamp=current.timestamp, + ) + if rss_growth_mb >= rss_growth_warning_mb: + return MemoryPressureEvent( + level="warning", + reason="rss_growth", + rss_mb=current.rss_mb, + threshold_mb=rss_growth_warning_mb, + rss_growth_mb=round(rss_growth_mb, 2), + window_s=round(window_s, 1), + timestamp=current.timestamp, + ) + return None + + +def _get_state_value(state) -> str: + return state.value if hasattr(state, "value") else str(state) + + +def _get_all_snapshots() -> dict: + try: + from yuxi.channel.runtime.manager import gateway # 延迟导入以避免循环依赖；Python import 缓存保证无重复加载开销 + + return gateway.get_all_snapshots() + except Exception: + logger.warning("DiagnosticHeartbeat: get_all_snapshots failed", exc_info=True) + return {} + + +def _count_active_sessions_from(snapshots: dict) -> int: + return sum(1 for s in snapshots.values() if _get_state_value(s.state) == "running") + + +diagnostic_heartbeat = DiagnosticHeartbeat() diff --git a/backend/package/yuxi/channel/monitoring/diagnostic_phase.py b/backend/package/yuxi/channel/monitoring/diagnostic_phase.py new file mode 100644 index 00000000..348e5def --- /dev/null +++ b/backend/package/yuxi/channel/monitoring/diagnostic_phase.py @@ -0,0 +1,131 @@ +import logging +import math +import os +import time +from contextlib import asynccontextmanager +from dataclasses import dataclass + +logger = logging.getLogger(__name__) + +RECENT_PHASE_CAPACITY = 40 + + +@dataclass +class PhaseSnapshot: + name: str + started_at: float + ended_at: float + duration_ms: float + cpu_user_ms: float + cpu_system_ms: float + cpu_total_ms: float + cpu_core_ratio: float + details: dict[str, str | int | float | bool] | None = None + + +_active_phase_stack: list[dict] = [] +_recent_phases: list[PhaseSnapshot] = [] + + +def _round_metric(value: float, digits: int = 1) -> float: + if not math.isfinite(value): + return 0.0 + factor = 10**digits + return round(value * factor) / factor + + +def _push_recent(snapshot: PhaseSnapshot) -> None: + _recent_phases.append(snapshot) + if len(_recent_phases) > RECENT_PHASE_CAPACITY: + del _recent_phases[: len(_recent_phases) - RECENT_PHASE_CAPACITY] + + +def _record_phase(snapshot: PhaseSnapshot) -> None: + _push_recent(snapshot) + try: + from yuxi.channel.monitoring.diagnostic_stability import record_stability_event + + record_stability_event( + "diagnostic.phase.completed", + name=snapshot.name, + started_at=snapshot.started_at, + ended_at=snapshot.ended_at, + duration_ms=snapshot.duration_ms, + cpu_user_ms=snapshot.cpu_user_ms, + cpu_system_ms=snapshot.cpu_system_ms, + cpu_total_ms=snapshot.cpu_total_ms, + cpu_core_ratio=snapshot.cpu_core_ratio, + ) + except Exception: + pass + + +def get_current_phase() -> str | None: + if not _active_phase_stack: + return None + return _active_phase_stack[-1]["name"] + + +def get_recent_phases(limit: int = 8) -> list[PhaseSnapshot]: + return list(_recent_phases[-max(0, limit) :]) + + +def record_diagnostic_phase(snapshot: PhaseSnapshot) -> None: + _record_phase(snapshot) + + +def reset_diagnostic_phases_for_test() -> None: + _active_phase_stack.clear() + _recent_phases.clear() + + +@asynccontextmanager +async def diagnostic_phase(name: str, details: dict[str, str | int | float | bool] | None = None): + cpu_started = os.times() + active: dict = { + "name": name, + "started_at": time.time(), + "started_wall": time.perf_counter(), + "cpu_started_user": cpu_started.user, + "cpu_started_system": cpu_started.system, + "details": details, + } + _active_phase_stack.append(active) + try: + yield + finally: + ended_at = time.time() + duration_ms = _round_metric((time.perf_counter() - active["started_wall"]) * 1000) + + cpu_ended = os.times() + cpu_user_ms = _round_metric((cpu_ended.user - active["cpu_started_user"]) * 1000) + cpu_system_ms = _round_metric((cpu_ended.system - active["cpu_started_system"]) * 1000) + cpu_total_ms = _round_metric(cpu_user_ms + cpu_system_ms) + + cpu_core_ratio = _round_metric(cpu_total_ms / max(1.0, duration_ms), 3) + + _active_phase_stack[:] = [e for e in _active_phase_stack if e is not active] + + snapshot = PhaseSnapshot( + name=name, + started_at=active["started_at"], + ended_at=ended_at, + duration_ms=duration_ms, + cpu_user_ms=cpu_user_ms, + cpu_system_ms=cpu_system_ms, + cpu_total_ms=cpu_total_ms, + cpu_core_ratio=cpu_core_ratio, + details=active["details"], + ) + _record_phase(snapshot) + + if cpu_core_ratio > 0.9: + logger.debug( + "DiagnosticPhase '%s': %sms, cpu_user=%.1fms, cpu_system=%.1fms, cpu_total=%.1fms, ratio=%.2f", + name, + duration_ms, + cpu_user_ms, + cpu_system_ms, + cpu_total_ms, + cpu_core_ratio, + ) diff --git a/backend/package/yuxi/channel/monitoring/diagnostic_stability.py b/backend/package/yuxi/channel/monitoring/diagnostic_stability.py new file mode 100644 index 00000000..4f7976e4 --- /dev/null +++ b/backend/package/yuxi/channel/monitoring/diagnostic_stability.py @@ -0,0 +1,409 @@ +import json +import logging +import os +import re +import signal +import sys +import time +from collections import deque +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from yuxi.channel.security.log_sanitizer import sanitize_text + +logger = logging.getLogger(__name__) + +DEFAULT_RING_CAPACITY = 1000 +DEFAULT_QUERY_LIMIT = 50 +MAX_BUNDLE_BYTES = 5 * 1024 * 1024 +DEFAULT_BUNDLE_RETENTION = 20 +MAX_SAFE_ERROR_MESSAGE_LENGTH = 500 + +SAFE_REASON_CODE = re.compile(r"^[A-Za-z0-9_.:-]{1,120}$") +REDACTED_HOSTNAME = "" + +_PEM_PRIVATE_KEY_PATTERN = re.compile( + r"-----BEGIN\s+(?:RSA\s+)?PRIVATE\s+KEY-----[\s\S]*?" + r"-----END\s+(?:RSA\s+)?PRIVATE\s+KEY-----", +) + + +@dataclass +class StabilityEvent: + seq: int + ts: float + type: str + data: dict[str, Any] = field(default_factory=dict) + + +class StabilityRingBuffer: + def __init__(self, capacity: int = DEFAULT_RING_CAPACITY): + self._buffer: deque[StabilityEvent] = deque(maxlen=capacity) + self._seq = 0 + self._dropped = 0 + + def record(self, event_type: str, **data: Any) -> None: + self._seq += 1 + event = StabilityEvent( + seq=self._seq, + ts=time.time(), + type=event_type, + data=data, + ) + if len(self._buffer) >= self._buffer.maxlen: + self._dropped += 1 + self._buffer.append(event) + + def query( + self, + limit: int = DEFAULT_QUERY_LIMIT, + event_type: str | None = None, + since_seq: int | None = None, + ) -> list[dict]: + events = list(self._buffer) + if event_type is not None: + events = [e for e in events if e.type == event_type] + if since_seq is not None: + events = [e for e in events if e.seq > since_seq] + events = events[-limit:] + return [ + { + "seq": e.seq, + "ts": e.ts, + "type": e.type, + **e.data, + } + for e in events + ] + + def snapshot( + self, + limit: int = DEFAULT_QUERY_LIMIT, + event_type: str | None = None, + since_seq: int | None = None, + ) -> dict: + all_events = list(self._buffer) + filtered = all_events + if event_type is not None: + filtered = [e for e in filtered if e.type == event_type] + if since_seq is not None: + filtered = [e for e in filtered if e.seq > since_seq] + events = [ + { + "seq": e.seq, + "ts": e.ts, + "type": e.type, + **e.data, + } + for e in filtered[-limit:] + ] + + by_type: dict[str, int] = {} + for e in filtered: + by_type[e.type] = by_type.get(e.type, 0) + 1 + + return { + "generated_at": time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime()), + "capacity": self._buffer.maxlen, + "count": len(filtered), + "dropped": self._dropped, + "first_seq": filtered[0].seq if filtered else None, + "last_seq": filtered[-1].seq if filtered else None, + "summary": {"by_type": by_type}, + "events": events, + } + + @property + def count(self) -> int: + return len(self._buffer) + + +_stability_buffer = StabilityRingBuffer() + +_pending_stability_reason: str | None = None +_pending_stability_error: Exception | None = None + + +def _write_stability_at_exit() -> None: + global _pending_stability_reason, _pending_stability_error + if _pending_stability_reason: + try: + write_stability_bundle_for_failure(_pending_stability_reason, _pending_stability_error) + except Exception: + pass + _pending_stability_reason = None + _pending_stability_error = None + + +def record_stability_event(event_type: str, **data: Any) -> None: + _stability_buffer.record(event_type, **data) + + +def get_stability_snapshot( + limit: int = DEFAULT_QUERY_LIMIT, + event_type: str | None = None, + since_seq: int | None = None, +) -> dict: + return _stability_buffer.snapshot(limit, event_type=event_type, since_seq=since_seq) + + +def _normalize_reason(reason: str) -> str: + return reason if SAFE_REASON_CODE.match(reason) else "unknown" + + +def _redact_sensitive_text(text: str) -> str: + text = sanitize_text(text) + text = _PEM_PRIVATE_KEY_PATTERN.sub("[REDACTED]", text) + return re.sub(r"\s+", " ", text).strip() + + +def _extract_error_name(error: Exception) -> str | None: + name = getattr(error, "name", None) + if isinstance(name, str) and SAFE_REASON_CODE.match(name): + return name + cls_name = type(error).__name__ + return cls_name if SAFE_REASON_CODE.match(cls_name) else None + + +def _extract_error_code(error: Exception) -> str | None: + code = getattr(error, "code", None) + if isinstance(code, str) and SAFE_REASON_CODE.match(code): + return code + if isinstance(code, int) and code >= 0: + return str(code) + errno = getattr(error, "errno", None) + if isinstance(errno, int) and errno >= 0: + return str(errno) + return None + + +def _extract_safe_error_message(error: Exception) -> str | None: + message = getattr(error, "message", None) or str(error) + if not message: + return None + sanitized = _redact_sensitive_text(message) + if not sanitized: + return None + if len(sanitized) > MAX_SAFE_ERROR_MESSAGE_LENGTH: + sanitized = sanitized[:MAX_SAFE_ERROR_MESSAGE_LENGTH] + "..." + return sanitized + + +def _build_error_metadata(error: Exception) -> dict[str, str] | None: + name = _extract_error_name(error) + code = _extract_error_code(error) + message = _extract_safe_error_message(error) + if not name and not code and not message: + return None + result: dict[str, str] = {} + if name: + result["name"] = name + if code: + result["code"] = code + if message: + result["message"] = message + return result + + +def _resolve_bundle_dir() -> Path: + return Path(os.environ.get("YUXI_STABILITY_DIR", "data/stability")) + + +def _get_hostname_redacted() -> str: + return REDACTED_HOSTNAME + + +def _get_process_uptime_ms() -> int: + try: + return int((time.time() - __import__("psutil").Process(os.getpid()).create_time()) * 1000) + except Exception: + return 0 + + +def _build_bundle_filename(timestamp: str, pid: int, reason: str) -> str: + normalized = _normalize_reason(reason) + return f"yuxi-stability-{timestamp}-{pid}-{normalized}.json" + + +def _truncate_events_to_fit(bundle: dict, snapshot: dict, max_bytes: int) -> dict: + events = snapshot["events"] + if not events: + return bundle + + lo, hi = 0, len(events) + while lo < hi: + mid = (lo + hi) // 2 + test_snapshot = {**snapshot, "events": events[mid:]} + test_bundle = {**bundle, "snapshot": test_snapshot} + if len(json.dumps(test_bundle, ensure_ascii=False, indent=2).encode()) <= max_bytes: + hi = mid + else: + lo = mid + 1 + + snapshot["events"] = events[lo:] + bundle["snapshot"] = snapshot + return bundle + + +def write_stability_bundle( + reason: str, + error: Exception | None = None, + include_empty: bool = False, + limit: int = DEFAULT_RING_CAPACITY, +) -> str | None: + import platform + + snapshot = _stability_buffer.snapshot(limit) + if snapshot["count"] == 0 and not error and not include_empty: + return None + + bundle_dir = _resolve_bundle_dir() + bundle_dir.mkdir(parents=True, exist_ok=True) + + normalized_reason = _normalize_reason(reason) + error_meta = _build_error_metadata(error) if error else None + + bundle = { + "version": 1, + "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "reason": normalized_reason, + "process": { + "pid": os.getpid(), + "platform": sys.platform, + "arch": platform.machine(), + "python": sys.version, + "uptime_ms": _get_process_uptime_ms(), + }, + "host": { + "hostname": _get_hostname_redacted(), + }, + "error": error_meta, + "snapshot": snapshot, + } + + timestamp = time.strftime("%Y%m%dT%H%M%S", time.localtime()) + filename = _build_bundle_filename(timestamp, os.getpid(), normalized_reason) + filepath = bundle_dir / filename + + content = json.dumps(bundle, ensure_ascii=False, indent=2) + content_bytes = content.encode() + if len(content_bytes) > MAX_BUNDLE_BYTES: + logger.error("Stability bundle too large (%d bytes), truncating", len(content_bytes)) + bundle = _truncate_events_to_fit(bundle, snapshot, MAX_BUNDLE_BYTES) + content = json.dumps(bundle, ensure_ascii=False, indent=2) + + with open(filepath, "w", encoding="utf-8") as f: + f.write(content) + + _restrict_file_permissions(filepath) + + _prune_old_bundles(bundle_dir) + logger.warning("Stability bundle written: %s", filepath) + return str(filepath) + + +def write_stability_bundle_for_failure(reason: str, error: Exception | None = None) -> str | None: + return write_stability_bundle(reason, error=error, include_empty=True) + + +def _restrict_file_permissions(filepath: Path) -> None: + try: + os.chmod(filepath, 0o600) + except Exception: + pass + + +def _prune_old_bundles(bundle_dir: Path) -> None: + try: + bundles = sorted( + bundle_dir.glob("yuxi-stability-*.json"), + key=lambda p: p.stat().st_mtime, + reverse=True, + ) + for p in bundles[DEFAULT_BUNDLE_RETENTION:]: + try: + p.unlink() + except OSError: + pass + except Exception: + pass + + +def list_stability_bundles(bundle_dir: Path | None = None) -> list[dict]: + _dir = bundle_dir or _resolve_bundle_dir() + try: + bundles = sorted( + _dir.glob("yuxi-stability-*.json"), + key=lambda p: p.stat().st_mtime, + reverse=True, + ) + return [{"path": str(p), "mtime": p.stat().st_mtime} for p in bundles] + except Exception: + return [] + + +def read_stability_bundle_file(filepath: str | Path) -> dict | None: + filepath = Path(filepath) + try: + stat = filepath.stat() + if stat.st_size > MAX_BUNDLE_BYTES: + logger.error("Stability bundle too large: %s (%d bytes)", filepath, stat.st_size) + return None + + with open(filepath, encoding="utf-8") as f: + raw = f.read() + + bundle = json.loads(raw) + if bundle.get("version") != 1: + logger.error("Unsupported stability bundle version: %s", bundle.get("version")) + return None + + reason = bundle.get("reason", "") + if not SAFE_REASON_CODE.match(reason): + bundle["reason"] = "unknown" + + host = bundle.get("host", {}) + if isinstance(host, dict): + bundle["host"]["hostname"] = REDACTED_HOSTNAME + + return bundle + except (json.JSONDecodeError, OSError) as e: + logger.error("Failed to read stability bundle %s: %s", filepath, e) + return None + + +def read_latest_stability_bundle(bundle_dir: Path | None = None) -> dict | None: + bundles = list_stability_bundles(bundle_dir) + if not bundles: + return None + return read_stability_bundle_file(bundles[0]["path"]) + + +_original_excepthook = sys.excepthook + + +def install_stability_crash_hook() -> None: + import atexit + + atexit.register(_write_stability_at_exit) + + def _handler(signum, frame): + global _pending_stability_reason + _pending_stability_reason = f"signal-{signum}" + signal.signal(signum, signal.SIG_DFL) + os.kill(os.getpid(), signum) + + for sig in (signal.SIGTERM, signal.SIGINT): + try: + signal.signal(sig, _handler) + except Exception: + pass + + def _excepthook(exc_type, exc_value, exc_tb): + global _pending_stability_reason, _pending_stability_error + _pending_stability_reason = "unhandled-exception" + _pending_stability_error = exc_value + _original_excepthook(exc_type, exc_value, exc_tb) + + sys.excepthook = _excepthook diff --git a/backend/package/yuxi/channel/monitoring/event_loop_monitor.py b/backend/package/yuxi/channel/monitoring/event_loop_monitor.py new file mode 100644 index 00000000..bc3dd2b2 --- /dev/null +++ b/backend/package/yuxi/channel/monitoring/event_loop_monitor.py @@ -0,0 +1,249 @@ +import asyncio +import bisect +import logging +import os +import time +from collections import deque +from dataclasses import dataclass, field + +logger = logging.getLogger(__name__) + +DEFAULT_SAMPLE_INTERVAL_MS = 100 +DEFAULT_WINDOW_SIZE = 600 +WARN_DELAY_MS = 1000 +CRITICAL_DELAY_MS = 5000 +WARN_UTIL = 0.95 +CPU_CORE_RATIO_WARN = 0.9 +DELAY_COINCIDENCE_MS = 25 +SUSTAINED_LOAD_MIN_INTERVAL_MS = 1000 + + +@dataclass +class EventLoopStats: + p50_ms: float = 0.0 + p90_ms: float = 0.0 + p99_ms: float = 0.0 + p999_ms: float = 0.0 + max_ms: float = 0.0 + min_ms: float = 0.0 + avg_ms: float = 0.0 + utilization: float = 0.0 + samples: int = 0 + warnings: list[str] = field(default_factory=list) + + +@dataclass +class EventLoopHealth: + degraded: bool = False + reasons: list[str] = field(default_factory=list) + interval_ms: float = 0.0 + delay_p99_ms: float = 0.0 + delay_max_ms: float = 0.0 + utilization: float = 0.0 + cpu_core_ratio: float = 0.0 + + +class EventLoopMonitor: + """监控 asyncio event loop 延迟，提供 P50/P90/P99/P999 延迟 + 利用率 + CPU。 + + 通过定期调度回调测量 event loop 处理延迟。 + 等效于 Node.js perf_hooks.monitorEventLoopDelay。 + """ + + def __init__( + self, + sample_interval_ms: int = DEFAULT_SAMPLE_INTERVAL_MS, + window_size: int = DEFAULT_WINDOW_SIZE, + ): + self._sample_interval = sample_interval_ms / 1000.0 + self._window_size = window_size + self._samples: list[float] = [] + self._ring: deque[float] = deque() + self._running = False + self._task: asyncio.Task | None = None + self._last_wall_at: float = 0.0 + self._last_process_time: float = 0.0 + self._cpu_count = _cpu_core_count() + self._last_health: EventLoopHealth | None = None + + async def start(self): + self._running = True + self._last_wall_at = time.perf_counter() + self._last_process_time = time.process_time() + self._task = asyncio.create_task(self._loop()) + logger.info( + "EventLoopMonitor started (interval=%dms, window=%d)", + int(self._sample_interval * 1000), + self._window_size, + ) + + async def stop(self): + self._running = False + if self._task: + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + logger.info("EventLoopMonitor stopped") + + async def _loop(self): + while self._running: + t0 = time.perf_counter() + await asyncio.sleep(self._sample_interval) + delay_ms = max(0.0, (time.perf_counter() - t0 - self._sample_interval) * 1000) + + self._ring.append(delay_ms) + bisect.insort(self._samples, delay_ms) + + if len(self._ring) > self._window_size: + evicted = self._ring.popleft() + idx = bisect.bisect_left(self._samples, evicted) + self._samples.pop(idx) + + def stats(self) -> EventLoopStats: + if not self._samples: + return EventLoopStats(samples=0) + + sorted_samples = self._samples + n = len(sorted_samples) + + def _pct(p: float) -> float: + idx = max(0, min(n - 1, int(n * p / 100.0))) + return sorted_samples[idx] + + p50 = _pct(50) + p99 = _pct(99) + avg = sum(sorted_samples) / n + util = p99 / self._sample_interval / 1000.0 if self._sample_interval > 0 else 0.0 + + warnings: list[str] = [] + if p99 > WARN_DELAY_MS: + warnings.append(f"P99延迟 {p99:.1f}ms > {WARN_DELAY_MS}ms") + if util > WARN_UTIL: + warnings.append(f"利用率 {util:.1%} > {WARN_UTIL:.0%}") + if p99 > CRITICAL_DELAY_MS: + warnings.append(f"P99延迟 {p99:.1f}ms > {CRITICAL_DELAY_MS}ms (严重)") + + return EventLoopStats( + p50_ms=round(p50, 2), + p90_ms=round(_pct(90), 2), + p99_ms=round(p99, 2), + p999_ms=round(_pct(99.9), 2), + max_ms=round(sorted_samples[-1], 2), + min_ms=round(sorted_samples[0], 2), + avg_ms=round(avg, 2), + utilization=round(util, 4), + samples=n, + warnings=warnings, + ) + + @property + def last_health(self) -> EventLoopHealth | None: + return self._last_health + + def health(self) -> EventLoopHealth: + """评估 event loop 当前健康状态。 + + 注意：此方法会更新内部时间戳和 CPU 计数器，作为后续调用的基准。 + 如需只读访问上次评估结果，请使用 last_health 属性。 + """ + now = time.perf_counter() + interval_ms = max(1.0, (now - self._last_wall_at) * 1000) + stats = self.stats() + + if stats.samples == 0: + return EventLoopHealth() + + delay_p99_ms = stats.p99_ms + delay_max_ms = stats.max_ms + has_delay_warning = delay_p99_ms >= WARN_DELAY_MS or delay_max_ms >= WARN_DELAY_MS + + if not has_delay_warning and interval_ms < SUSTAINED_LOAD_MIN_INTERVAL_MS: + if self._last_health is not None: + return self._last_health + return EventLoopHealth( + interval_ms=round(interval_ms, 1), + delay_p99_ms=delay_p99_ms, + delay_max_ms=delay_max_ms, + utilization=stats.utilization, + ) + + current_process_time = time.process_time() + cpu_delta = current_process_time - self._last_process_time + cpu_core_ratio = round(cpu_delta / max(interval_ms / 1000, 0.001), 3) + + reasons = _classify_health_reasons( + interval_ms=interval_ms, + delay_p99_ms=delay_p99_ms, + delay_max_ms=delay_max_ms, + utilization=stats.utilization, + cpu_core_ratio=cpu_core_ratio, + ) + + health = EventLoopHealth( + degraded=len(reasons) > 0, + reasons=reasons, + interval_ms=round(interval_ms, 1), + delay_p99_ms=delay_p99_ms, + delay_max_ms=delay_max_ms, + utilization=stats.utilization, + cpu_core_ratio=cpu_core_ratio, + ) + + self._last_wall_at = now + self._last_process_time = current_process_time + self._last_health = health + + _push_event_loop_metrics(health) + + return health + + +def _classify_health_reasons( + *, + interval_ms: float, + delay_p99_ms: float, + delay_max_ms: float, + utilization: float, + cpu_core_ratio: float, +) -> list[str]: + reasons: list[str] = [] + + if delay_p99_ms >= WARN_DELAY_MS or delay_max_ms >= WARN_DELAY_MS: + reasons.append("event_loop_delay") + + if interval_ms < SUSTAINED_LOAD_MIN_INTERVAL_MS: + return reasons + + has_delay_co_evidence = delay_p99_ms >= DELAY_COINCIDENCE_MS or delay_max_ms >= DELAY_COINCIDENCE_MS + if not has_delay_co_evidence: + return reasons + + if utilization >= WARN_UTIL: + reasons.append("event_loop_utilization") + if cpu_core_ratio >= CPU_CORE_RATIO_WARN: + reasons.append("cpu") + + return reasons + + +def _cpu_core_count() -> int: + try: + return os.cpu_count() or 1 + except Exception: + return 1 + + +def _push_event_loop_metrics(health: EventLoopHealth): + try: + from yuxi.channel.monitoring.metrics import event_loop_delay_p99, event_loop_utilization + + event_loop_delay_p99.set(health.delay_p99_ms) + event_loop_utilization.set(health.utilization) + except Exception: + pass + + +event_loop_monitor = EventLoopMonitor() diff --git a/backend/package/yuxi/channel/monitoring/event_loop_ready.py b/backend/package/yuxi/channel/monitoring/event_loop_ready.py new file mode 100644 index 00000000..70f7a71b --- /dev/null +++ b/backend/package/yuxi/channel/monitoring/event_loop_ready.py @@ -0,0 +1,139 @@ +import asyncio +import logging +import time +from dataclasses import dataclass + +logger = logging.getLogger(__name__) + +DEFAULT_MAX_WAIT_MS = 10_000 +DEFAULT_INTERVAL_MS = 1 +DEFAULT_DRIFT_THRESHOLD_MS = 200 +DEFAULT_CONSECUTIVE_READY_CHECKS = 2 + +MAX_SAFE_TIMEOUT_DELAY_MS = 2_147_483_647 + + +@dataclass +class EventLoopReadyResult: + ready: bool + elapsed_ms: float + max_drift_ms: float + checks: int + aborted: bool + + +def _resolve_positive_int(value: float | int | None, fallback: int) -> int: + if value is None: + return fallback + candidate = int(value) + return max(1, candidate) if candidate > 0 else fallback + + +def _resolve_safe_delay_ms(delay_ms: float, min_ms: int = 1) -> int: + candidate = _resolve_positive_int(delay_ms, min_ms) + return min(MAX_SAFE_TIMEOUT_DELAY_MS, candidate) + + +async def wait_for_event_loop_ready( + *, + max_wait_ms: float = DEFAULT_MAX_WAIT_MS, + interval_ms: float = DEFAULT_INTERVAL_MS, + drift_threshold_ms: float = DEFAULT_DRIFT_THRESHOLD_MS, + consecutive_ready_checks: int = DEFAULT_CONSECUTIVE_READY_CHECKS, + cancel_event: asyncio.Event | None = None, +) -> EventLoopReadyResult: + """等待事件循环就绪，通过 setTimeout 漂移检测判断。 + + 在 asyncio 中通过 call_later 调度回调测量实际延迟与预期延迟之间的漂移。 + 当连续 consecutive_ready_checks 次漂移在阈值内时认为事件循环已就绪。 + + 可通过 cancel_event 参数取消等待，取消后返回 aborted=True 的结果。 + """ + max_wait_ms_val = float(_resolve_safe_delay_ms(max_wait_ms)) + interval_ms_val = float(_resolve_positive_int(interval_ms, DEFAULT_INTERVAL_MS)) + drift_threshold_ms_val = float(_resolve_positive_int(drift_threshold_ms, DEFAULT_DRIFT_THRESHOLD_MS)) + consecutive_ready_checks_val = _resolve_positive_int(consecutive_ready_checks, DEFAULT_CONSECUTIVE_READY_CHECKS) + + started_at = time.perf_counter() + ready_checks = 0 + checks = 0 + max_drift_ms = 0.0 + + loop = asyncio.get_running_loop() + fut: asyncio.Future[EventLoopReadyResult] = loop.create_future() + timer_handle: asyncio.TimerHandle | None = None + settled = False + + def clear_timer(): + nonlocal timer_handle + if timer_handle is not None: + timer_handle.cancel() + timer_handle = None + + def finish(ready: bool, aborted: bool = False): + nonlocal settled, timer_handle + if settled: + return + settled = True + clear_timer() + elapsed_ms = max(0.0, (time.perf_counter() - started_at) * 1000) + fut.set_result( + EventLoopReadyResult( + ready=ready, + elapsed_ms=round(elapsed_ms, 1), + max_drift_ms=round(max_drift_ms, 2), + checks=checks, + aborted=aborted, + ) + ) + + def schedule_next(): + nonlocal ready_checks, checks, max_drift_ms, timer_handle + + remaining_ms = max_wait_ms_val - max(0.0, (time.perf_counter() - started_at) * 1000) + if remaining_ms <= 0: + finish(False) + return + + delay_s = min(interval_ms_val, remaining_ms) / 1000 + scheduled_at = time.perf_counter() + + def on_timer(): + nonlocal ready_checks, checks, max_drift_ms, timer_handle + checks += 1 + drift_ms = max(0.0, (time.perf_counter() - scheduled_at) * 1000 - (delay_s * 1000)) + max_drift_ms = max(max_drift_ms, drift_ms) + + if drift_ms > drift_threshold_ms_val: + ready_checks = 0 + else: + ready_checks += 1 + + if ready_checks >= consecutive_ready_checks_val: + finish(True) + return + + schedule_next() + + timer_handle = loop.call_later(delay_s, on_timer) + + schedule_next() + + cancel_watch_task: asyncio.Task | None = None + if cancel_event is not None: + if cancel_event.is_set(): + finish(False, True) + else: + + async def _watch_cancel(): + await cancel_event.wait() + finish(False, True) + + cancel_watch_task = asyncio.create_task(_watch_cancel()) + + try: + return await fut + finally: + clear_timer() + if cancel_watch_task is not None: + cancel_watch_task.cancel() diff --git a/backend/package/yuxi/channel/monitoring/health_monitor.py b/backend/package/yuxi/channel/monitoring/health_monitor.py new file mode 100644 index 00000000..37ae0c6e --- /dev/null +++ b/backend/package/yuxi/channel/monitoring/health_monitor.py @@ -0,0 +1,468 @@ +import asyncio +import logging +import time +from dataclasses import dataclass, field +from typing import TypedDict + +logger = logging.getLogger(__name__) + +DEFAULT_CHECK_INTERVAL = 300 +DEFAULT_STARTUP_GRACE = 60 +DEFAULT_CONNECTION_GRACE = 120 +DEFAULT_COOLDOWN_CYCLES = 2 +DEFAULT_MAX_RESTARTS_PER_HOUR = 10 +DEFAULT_STALE_EVENT_THRESHOLD = 1800 +BUSY_ACTIVITY_STALE_THRESHOLD = 1500 +RECONNECT_GAVE_UP_THRESHOLD = 10 + + +class ChannelSnapshotData(TypedDict, total=False): + enabled: bool | None + configured: bool | None + running: bool + connected: bool + busy: bool + active_runs: int + last_run_activity_at: float | None + last_transport_activity_at: float | None + last_start_at: float | None + reconnect_attempts: int + restart_pending: bool | None + state: str + + +@dataclass +class ChannelHealthEvaluation: + healthy: bool + reason: str + detail: str | None = None + + +@dataclass +class ChannelHealthEvent: + channel_type: str + account_id: str + from_state: str + to_state: str + timestamp: float = field(default_factory=time.time) + detail: str | None = None + + def to_dict(self) -> dict: + return { + "channel_type": self.channel_type, + "account_id": self.account_id, + "from_state": self.from_state, + "to_state": self.to_state, + "timestamp": self.timestamp, + "detail": self.detail, + } + + +@dataclass +class ChannelHealthMetrics: + checks_total: int = 0 + transitions: list[ChannelHealthEvent] = field(default_factory=list) + current_status: str = "unknown" + last_check_at: float = 0.0 + unhealthy_channels: list[str] = field(default_factory=list) + degraded_channels: list[str] = field(default_factory=list) + evaluation_details: dict[str, dict] = field(default_factory=dict) + + def to_dict(self) -> dict: + return { + "checks_total": self.checks_total, + "current_status": self.current_status, + "last_check_at": self.last_check_at, + "unhealthy_channels": self.unhealthy_channels, + "degraded_channels": self.degraded_channels, + "evaluation_details": self.evaluation_details, + "recent_transitions": [t.to_dict() for t in self.transitions[-20:]], + } + + +def _safe_finite_timestamp(value: float | None) -> float | None: + if value is None: + return None + if not isinstance(value, (int, float)) or value != value: + return None + if value == float("inf") or value == float("-inf"): + return None + return value + + +def _safe_active_runs(value: object) -> int: + if isinstance(value, (int, float)): + if value != value or value == float("inf") or value == float("-inf"): + return 0 + return max(0, int(value)) + return 0 + + +def evaluate_channel_health( + *, + enabled: bool | None = None, + configured: bool | None = None, + running: bool = False, + connected: bool = False, + busy: bool = False, + active_runs: int = 0, + last_run_activity_at: float | None = None, + last_transport_activity_at: float | None = None, + last_start_at: float | None = None, + reconnect_attempts: int = 0, + now: float | None = None, + stale_event_threshold: float = DEFAULT_STALE_EVENT_THRESHOLD, + connect_grace: float = DEFAULT_CONNECTION_GRACE, +) -> ChannelHealthEvaluation: + if now is None: + now = time.monotonic() + + if enabled is False or configured is False: + return ChannelHealthEvaluation(healthy=True, reason="unmanaged") + + if not running: + reason = "gave-up" if reconnect_attempts >= RECONNECT_GAVE_UP_THRESHOLD else "not-running" + return ChannelHealthEvaluation( + healthy=False, + reason=reason, + detail=f"reconnect_attempts={reconnect_attempts}", + ) + + safe_active_runs = _safe_active_runs(active_runs) + is_busy = busy or safe_active_runs > 0 + + safe_last_start_at = _safe_finite_timestamp(last_start_at) + safe_last_run_activity_at = _safe_finite_timestamp(last_run_activity_at) + safe_last_transport_activity_at = _safe_finite_timestamp(last_transport_activity_at) + + if is_busy: + busy_state_initialized_for_lifecycle = safe_last_start_at is None or ( + safe_last_run_activity_at is not None and safe_last_run_activity_at >= safe_last_start_at + ) + if busy_state_initialized_for_lifecycle: + run_activity_age = ( + float("inf") if safe_last_run_activity_at is None else max(0.0, now - safe_last_run_activity_at) + ) + if run_activity_age < BUSY_ACTIVITY_STALE_THRESHOLD: + return ChannelHealthEvaluation(healthy=True, reason="busy") + return ChannelHealthEvaluation( + healthy=False, + reason="stuck", + detail=f"run_activity_age={run_activity_age:.0f}s", + ) + + if safe_last_start_at is not None: + up_duration = now - safe_last_start_at + if up_duration < connect_grace: + return ChannelHealthEvaluation(healthy=True, reason="startup-connect-grace") + + if connected is False: + return ChannelHealthEvaluation(healthy=False, reason="disconnected") + + if connected and safe_last_transport_activity_at is not None: + if safe_last_start_at is not None and safe_last_transport_activity_at < safe_last_start_at: + lifecycle_gap = max(0.0, now - safe_last_start_at) + if lifecycle_gap <= stale_event_threshold: + return ChannelHealthEvaluation(healthy=True, reason="healthy") + return ChannelHealthEvaluation(healthy=False, reason="stale-socket") + + event_age = now - safe_last_transport_activity_at + if event_age > stale_event_threshold: + return ChannelHealthEvaluation( + healthy=False, + reason="stale-socket", + detail=f"event_age={event_age:.0f}s", + ) + + return ChannelHealthEvaluation(healthy=True, reason="healthy") + + +def resolve_restart_reason( + snapshot: ChannelSnapshotData, + evaluation: ChannelHealthEvaluation, +) -> str: + if evaluation.reason == "stale-socket": + return "stale-socket" + if evaluation.reason == "not-running": + reconnect_attempts = snapshot.get("reconnect_attempts", 0) + return "gave-up" if reconnect_attempts >= RECONNECT_GAVE_UP_THRESHOLD else "stopped" + if evaluation.reason == "disconnected": + return "disconnected" + if evaluation.reason == "stuck": + return "stuck" + return evaluation.reason + + +@dataclass +class _RestartRecord: + last_restart_at: float = 0.0 + restarts_this_hour: list[float] = field(default_factory=list) + + def prune(self, now: float) -> None: + cutoff = now - 3600 + self.restarts_this_hour = [t for t in self.restarts_this_hour if t > cutoff] + + +class ChannelHealthMonitor: + """频道健康自动监控器：独立后台轮询，5min 间隔自动检查。 + + 特性： + - 启动宽限 60s（启动期间不告警） + - 连接宽限 120s（新建连接等待稳定） + - 冷却 2 周期（时间窗口冷却，避免抖动恢复触发频繁重启） + - 每 channel 每小时重启上限 10 次（独立计数，不互相影响） + - Stale-socket / stuck / gave-up 检测 + - 检测到不健康的 channel 后自动重启（按 stop → start 流程） + - 自动跳过手动停止的 channel（不做重启） + """ + + def __init__( + self, + check_interval: int = DEFAULT_CHECK_INTERVAL, + startup_grace: int = DEFAULT_STARTUP_GRACE, + connection_grace: int = DEFAULT_CONNECTION_GRACE, + cooldown_cycles: int = DEFAULT_COOLDOWN_CYCLES, + max_restarts_per_hour: int = DEFAULT_MAX_RESTARTS_PER_HOUR, + stale_event_threshold: int = DEFAULT_STALE_EVENT_THRESHOLD, + ): + self._check_interval = check_interval + self._startup_grace = startup_grace + self._connection_grace = connection_grace + self._cooldown_cycles = cooldown_cycles + self._max_restarts_per_hour = max_restarts_per_hour + self._stale_event_threshold = stale_event_threshold + self._running = False + self._check_in_flight = False + self._task: asyncio.Task | None = None + self._started_at: float = 0.0 + self._previous_states: dict[str, str] = {} + self._restart_records: dict[str, _RestartRecord] = {} + self._metrics = ChannelHealthMetrics() + + @property + def metrics(self) -> ChannelHealthMetrics: + return self._metrics + + async def start(self): + self._running = True + self._started_at = time.monotonic() + self._task = asyncio.create_task(self._loop()) + logger.info( + "ChannelHealthMonitor started (interval=%ds, startup_grace=%ds, " + "connection_grace=%ds, cooldown=%d cycles, max_restarts=%d/h/channel, " + "stale_event_threshold=%ds)", + self._check_interval, + self._startup_grace, + self._connection_grace, + self._cooldown_cycles, + self._max_restarts_per_hour, + self._stale_event_threshold, + ) + + async def stop(self): + self._running = False + if self._task: + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + logger.info("ChannelHealthMonitor stopped") + + async def _loop(self): + while self._running: + await asyncio.sleep(self._check_interval) + try: + await self._check_and_alert() + except Exception: + logger.exception("ChannelHealthMonitor: check failed") + + async def _check_and_alert(self): + if self._check_in_flight: + return + self._check_in_flight = True + + try: + await self._do_check() + finally: + self._check_in_flight = False + + async def _do_check(self): + now = time.monotonic() + elapsed = now - self._started_at + + if elapsed < self._startup_grace: + logger.debug( + "ChannelHealthMonitor: startup grace period (%.0fs / %ds)", + elapsed, + self._startup_grace, + ) + return + + from yuxi.channel.runtime.manager import gateway # 延迟导入以避免循环依赖；Python import 缓存保证无重复加载开销 + + report = gateway.get_health() + self._metrics.checks_total += 1 + self._metrics.last_check_at = now + self._metrics.current_status = report.status + + cooldown_window = self._cooldown_cycles * self._check_interval + + unhealthy: list[str] = [] + degraded: list[str] = [] + evaluation_details: dict[str, dict] = {} + + for key, snap_data in report.channels.items(): + parts = key.split(":", 1) + if len(parts) != 2: + logger.warning("ChannelHealthMonitor: invalid channel key format: %s", key) + continue + channel_type, account_id = parts + state = snap_data.get("state", "stopped") + prev_state = self._previous_states.get(key, "stopped") + + evaluation = evaluate_channel_health( + enabled=snap_data.get("enabled"), + configured=snap_data.get("configured"), + running=snap_data.get("running", False), + connected=snap_data.get("connected", False), + busy=snap_data.get("busy", False), + active_runs=snap_data.get("active_runs", 0), + last_run_activity_at=snap_data.get("last_run_activity_at"), + last_transport_activity_at=snap_data.get("last_transport_activity_at"), + last_start_at=snap_data.get("last_start_at"), + reconnect_attempts=snap_data.get("reconnect_attempts", 0), + now=now, + stale_event_threshold=self._stale_event_threshold, + connect_grace=self._connection_grace, + ) + + evaluation_details[key] = { + "healthy": evaluation.healthy, + "reason": evaluation.reason, + "detail": evaluation.detail, + "restart_reason": resolve_restart_reason(snap_data, evaluation) if not evaluation.healthy else None, + } + + if prev_state != state: + event = ChannelHealthEvent( + channel_type=channel_type, + account_id=account_id, + from_state=prev_state, + to_state=state, + detail=snap_data.get("last_error"), + ) + self._metrics.transitions.append(event) + self._previous_states[key] = state + + logger.info( + "ChannelHealthMonitor: %s:%s %s → %s (health=%s)", + channel_type, + account_id, + prev_state, + state, + evaluation.reason, + ) + + if not evaluation.healthy and evaluation.reason not in ("unmanaged", "startup-connect-grace"): + unhealthy.append(key) + elif evaluation.reason == "startup-connect-grace": + degraded.append(key) + + if evaluation.healthy or evaluation.reason in ( + "startup-connect-grace", + "unmanaged", + ): + continue + + if gateway.is_manually_stopped(channel_type, account_id): + logger.debug("ChannelHealthMonitor: %s 手动停止中，跳过自动恢复", key) + continue + + restart_reason = resolve_restart_reason(snap_data, evaluation) + logger.warning( + "ChannelHealthMonitor: %s 不健康 (reason=%s, restart_reason=%s)，准备自动恢复", + key, + evaluation.reason, + restart_reason, + ) + + record = self._restart_records.get(key) + if record is None: + record = _RestartRecord() + self._restart_records[key] = record + + if now - record.last_restart_at <= cooldown_window: + continue + + record.prune(now) + if len(record.restarts_this_hour) >= self._max_restarts_per_hour: + logger.warning( + "ChannelHealthMonitor: %s 已达到每小时重启上限 (%d/%d)，跳过本次自动恢复", + key, + len(record.restarts_this_hour), + self._max_restarts_per_hour, + ) + continue + + record.last_restart_at = now + record.restarts_this_hour.append(now) + + logger.info( + "ChannelHealthMonitor: %s 触发自动恢复 (reason=%s)", + key, + restart_reason, + ) + try: + if snap_data.get("running"): + await gateway.stop_channel(channel_type, account_id) + await gateway.start_channel(channel_type, account_id) + except Exception: + logger.exception("ChannelHealthMonitor: %s 自动恢复执行失败", key) + + self._metrics.unhealthy_channels = unhealthy + self._metrics.degraded_channels = degraded + self._metrics.evaluation_details = evaluation_details + + known_keys = set(report.channels.keys()) + for key in list(self._restart_records.keys()): + if key not in known_keys: + del self._restart_records[key] + + if unhealthy: + logger.error( + "ChannelHealthMonitor: %d unhealthy channels: %s", + len(unhealthy), + ", ".join(unhealthy), + ) + _maybe_send_alert(unhealthy, evaluation_details) + if degraded: + logger.warning( + "ChannelHealthMonitor: %d degraded channels: %s", + len(degraded), + ", ".join(degraded), + ) + + def record_restart(self, key: str): + record = self._restart_records.get(key) + if record is None: + record = _RestartRecord() + self._restart_records[key] = record + record.last_restart_at = time.monotonic() + record.restarts_this_hour.append(time.monotonic()) + + +async def _maybe_send_alert(unhealthy: list[str], details: dict): + try: + from yuxi.channel.monitoring.metrics import send_alert_webhook + + await send_alert_webhook( + f"ChannelHealthMonitor: {len(unhealthy)} unhealthy channels", + {"channels": unhealthy, "details": details}, + ) + except Exception: + pass + + +channel_health_monitor = ChannelHealthMonitor() diff --git a/backend/package/yuxi/channel/monitoring/metrics.py b/backend/package/yuxi/channel/monitoring/metrics.py new file mode 100644 index 00000000..49747db5 --- /dev/null +++ b/backend/package/yuxi/channel/monitoring/metrics.py @@ -0,0 +1,134 @@ +import logging +import os +import time +from collections import deque +from dataclasses import dataclass, field + +logger = logging.getLogger(__name__) + +HISTORY_CAPACITY = 60 + + +@dataclass +class _MetricPoint: + ts: float + value: float + labels: dict[str, str] = field(default_factory=dict) + + +class _MetricFamily: + def __init__(self, name: str, help_text: str, metric_type: str): + self.name = name + self.help = help_text + self.type = metric_type + self._buffer: deque[_MetricPoint] = deque(maxlen=HISTORY_CAPACITY) + self._last_value: float | None = None + self._last_labels: dict[str, str] | None = None + + def set(self, value: float, labels: dict[str, str] | None = None): + self._last_value = value + self._last_labels = labels + self._buffer.append(_MetricPoint(ts=time.time(), value=value, labels=labels or {})) + + @property + def value(self) -> float | None: + return self._last_value + + def history(self) -> list[dict]: + return [{"ts": p.ts, "value": p.value, "labels": p.labels} for p in self._buffer] + + def prometheus_line(self) -> str: + if self._last_value is None: + return "" + labels = self._last_labels or {} + label_str = "{" + ",".join(f'{k}="{v}"' for k, v in labels.items()) + "}" if labels else "" + return f"{self.name}{label_str} {self._last_value}\n" + + +class MetricsRegistry: + def __init__(self): + self._metrics: dict[str, _MetricFamily] = {} + + def gauge(self, name: str, help_text: str) -> _MetricFamily: + if name not in self._metrics: + self._metrics[name] = _MetricFamily(name, help_text, "gauge") + return self._metrics[name] + + def collect_prometheus(self) -> str: + lines = [] + seen_help: set[str] = set() + for m in self._metrics.values(): + if m.name not in seen_help: + lines.append(f"# HELP {m.name} {m.help}") + lines.append(f"# TYPE {m.name} {m.type}") + seen_help.add(m.name) + line = m.prometheus_line() + if line: + lines.append(line) + return "\n".join(lines) + "\n" + + def history(self, name: str) -> list[dict]: + m = self._metrics.get(name) + return m.history() if m else [] + + def all_history(self) -> dict[str, list[dict]]: + return {name: m.history() for name, m in self._metrics.items()} + + +metrics_registry = MetricsRegistry() + +event_loop_delay_p99 = metrics_registry.gauge( + "yuxi_event_loop_delay_p99_ms", + "Event loop P99 delay in milliseconds", +) +event_loop_utilization = metrics_registry.gauge( + "yuxi_event_loop_utilization", + "Event loop utilization ratio", +) +memory_rss_mb = metrics_registry.gauge( + "yuxi_memory_rss_mb", + "Process RSS memory in MB", +) +memory_gc_objects = metrics_registry.gauge( + "yuxi_memory_gc_objects", + "Number of objects tracked by GC", +) +active_sessions = metrics_registry.gauge( + "yuxi_active_sessions", + "Number of active channel sessions", +) +stuck_sessions = metrics_registry.gauge( + "yuxi_stuck_sessions", + "Number of stuck channel sessions", +) +channel_health = metrics_registry.gauge( + "yuxi_channel_health", + "Channel health status (1=healthy, 0=unhealthy)", +) +active_tasks = metrics_registry.gauge( + "yuxi_active_tasks", + "Number of active asyncio tasks", +) + + +async def send_alert_webhook(title: str, details: dict | None = None): + webhook_url = os.environ.get("YUXI_ALERT_WEBHOOK_URL") + if not webhook_url: + return + + payload = { + "title": title, + "timestamp": time.time(), + } + if details: + payload["details"] = details + + try: + import aiohttp + + async with aiohttp.ClientSession() as session: + async with session.post(webhook_url, json=payload, timeout=aiohttp.ClientTimeout(total=5)) as resp: + if resp.status >= 400: + logger.warning("Alert webhook failed: HTTP %s", resp.status) + except Exception: + logger.warning("Alert webhook delivery failed", exc_info=True) diff --git a/backend/package/yuxi/channel/monitoring/readiness.py b/backend/package/yuxi/channel/monitoring/readiness.py new file mode 100644 index 00000000..084fbf94 --- /dev/null +++ b/backend/package/yuxi/channel/monitoring/readiness.py @@ -0,0 +1,204 @@ +import logging +import time +from collections.abc import Callable +from dataclasses import dataclass, field + +from yuxi.channel.monitoring.event_loop_monitor import EventLoopHealth +from yuxi.channel.monitoring.health_monitor import ( + DEFAULT_CONNECTION_GRACE, + DEFAULT_STALE_EVENT_THRESHOLD, + evaluate_channel_health, +) + +logger = logging.getLogger(__name__) + +DEFAULT_READINESS_CACHE_MS = 1000.0 + + +@dataclass +class ChannelReadiness: + key: str + healthy: bool + reason: str + detail: str | None = None + + +@dataclass +class ReadinessResult: + ready: bool + channels: list[ChannelReadiness] + failing: list[str] = field(default_factory=list) + uptime_ms: float = 0.0 + event_loop_degraded: bool = False + event_loop_reasons: list[str] = field(default_factory=list) + event_loop: EventLoopHealth | None = None + checked_at: float = 0.0 + cache_ttl_ms: float = 1000.0 + + +def _should_ignore_readiness_failure( + evaluation_reason: str, + snap_data: dict, +) -> bool: + if evaluation_reason == "unmanaged": + return True + if evaluation_reason == "stale-socket": + return True + if evaluation_reason == "not-running" and snap_data.get("restart_pending") is True: + return True + return False + + +class ReadinessChecker: + def __init__( + self, + stale_event_threshold: float = DEFAULT_STALE_EVENT_THRESHOLD, + connect_grace: float = DEFAULT_CONNECTION_GRACE, + cache_ttl_ms: float = DEFAULT_READINESS_CACHE_MS, + started_at: float | None = None, + get_startup_pending: Callable[[], bool] | None = None, + get_startup_pending_reason: Callable[[], str] | None = None, + get_event_loop_health: Callable[[], EventLoopHealth | None] | None = None, + should_skip_channel_readiness: Callable[[], bool] | None = None, + ): + self._stale_event_threshold = stale_event_threshold + self._connect_grace = connect_grace + self._cache_ttl_ms = max(0.0, cache_ttl_ms) + self._started_at = started_at if started_at is not None else time.monotonic() + self._get_startup_pending = get_startup_pending + self._get_startup_pending_reason = get_startup_pending_reason + self._get_event_loop_health = get_event_loop_health + self._should_skip_channel_readiness = should_skip_channel_readiness + self._last_result: ReadinessResult | None = None + self._last_check_at: float = 0.0 + + async def check(self, *, force: bool = False) -> ReadinessResult: + now = time.monotonic() + uptime_ms = (now - self._started_at) * 1000 + + if self._get_startup_pending and self._get_startup_pending(): + reason = self._get_startup_pending_reason() if self._get_startup_pending_reason else "startup-sidecars" + result = ReadinessResult( + ready=False, + channels=[], + failing=[reason], + uptime_ms=uptime_ms, + checked_at=now, + cache_ttl_ms=self._cache_ttl_ms, + ) + result = self._attach_event_loop(result) + return result + + if self._should_skip_channel_readiness and self._should_skip_channel_readiness(): + result = ReadinessResult( + ready=True, + channels=[], + failing=[], + uptime_ms=uptime_ms, + checked_at=now, + cache_ttl_ms=self._cache_ttl_ms, + ) + result = self._attach_event_loop(result) + return result + + if not force and self._last_result is not None: + elapsed = (now - self._last_check_at) * 1000 + if elapsed < self._cache_ttl_ms: + cached = ReadinessResult( + ready=self._last_result.ready, + channels=list(self._last_result.channels), + failing=list(self._last_result.failing), + uptime_ms=uptime_ms, + event_loop_degraded=self._last_result.event_loop_degraded, + event_loop_reasons=list(self._last_result.event_loop_reasons), + event_loop=self._last_result.event_loop, + checked_at=self._last_result.checked_at, + cache_ttl_ms=self._last_result.cache_ttl_ms, + ) + cached = self._attach_event_loop(cached) + return cached + + channel_evaluations: list[ChannelReadiness] = [] + failing: list[str] = [] + + try: + from yuxi.channel.runtime.manager import gateway + + report = gateway.get_health() + for key, snap_data in report.channels.items(): + evaluation = evaluate_channel_health( + enabled=snap_data.get("enabled"), + configured=snap_data.get("configured"), + running=snap_data.get("running", False), + connected=snap_data.get("connected", False), + busy=snap_data.get("busy", False), + active_runs=snap_data.get("active_runs", 0), + last_run_activity_at=snap_data.get("last_run_activity_at"), + last_transport_activity_at=snap_data.get("last_transport_activity_at"), + last_start_at=snap_data.get("last_start_at"), + reconnect_attempts=snap_data.get("reconnect_attempts", 0), + now=now, + stale_event_threshold=self._stale_event_threshold, + connect_grace=self._connect_grace, + ) + if evaluation.reason == "unmanaged": + continue + + effective_healthy = ( + True + if not evaluation.healthy and _should_ignore_readiness_failure(evaluation.reason, snap_data) + else evaluation.healthy + ) + + channel_evaluations.append( + ChannelReadiness( + key=key, + healthy=effective_healthy, + reason=evaluation.reason, + detail=evaluation.detail, + ) + ) + + if not effective_healthy: + failing.append(key) + except Exception as e: + logger.warning("ReadinessChecker: failed to evaluate channels: %s", e) + failing.append("evaluation-error") + + all_ready = len(failing) == 0 + + result = ReadinessResult( + ready=all_ready, + channels=channel_evaluations, + failing=failing, + uptime_ms=uptime_ms, + checked_at=now, + cache_ttl_ms=self._cache_ttl_ms, + ) + result = self._attach_event_loop(result) + if result.event_loop_degraded: + all_ready = False + result.ready = all_ready + + self._last_result = result + self._last_check_at = now + return result + + def _attach_event_loop(self, result: ReadinessResult) -> ReadinessResult: + try: + if self._get_event_loop_health: + event_loop = self._get_event_loop_health() + else: + from yuxi.channel.monitoring.event_loop_monitor import event_loop_monitor + + event_loop = event_loop_monitor.health() + if event_loop is not None: + result.event_loop = event_loop + result.event_loop_degraded = event_loop.degraded + result.event_loop_reasons = list(event_loop.reasons) + except Exception as e: + logger.debug("ReadinessChecker: failed to get event loop health: %s", e) + return result + + +readiness_checker = ReadinessChecker() diff --git a/backend/package/yuxi/channel/monitoring/state_aggregator.py b/backend/package/yuxi/channel/monitoring/state_aggregator.py new file mode 100644 index 00000000..33de9051 --- /dev/null +++ b/backend/package/yuxi/channel/monitoring/state_aggregator.py @@ -0,0 +1,233 @@ +from dataclasses import dataclass, field +from typing import Any + + +@dataclass +class ChannelHealthState: + channel_type: str + account_id: str + state: str = "" + connected: bool = False + healthy: bool = True + health_reason: str = "unknown" + health_detail: str | None = None + last_event_at: float = 0.0 + last_error: str | None = None + active_runs: int = 0 + uptime_seconds: float = 0.0 + restart_count: int = 0 + + +@dataclass +class ConfigHealthState: + status: str = "healthy" + last_known_good: str | None = None + issues: list[str] = field(default_factory=list) + revision: int = 0 + + +@dataclass +class MonitoringState: + checks_total: int = 0 + current_status: str = "unknown" + unhealthy_channels: list[str] = field(default_factory=list) + degraded_channels: list[str] = field(default_factory=list) + last_check_at: float = 0.0 + + +@dataclass +class ReadinessState: + ready: bool = False + failing: list[str] = field(default_factory=list) + uptime_ms: float = 0.0 + event_loop_degraded: bool = False + + +@dataclass +class DiagnosticState: + overall_status: str = "pass" + timestamp: float = 0.0 + warnings_count: int = 0 + security_warnings_count: int = 0 + + +@dataclass +class GatewayConnectionState: + active_connections: int = 0 + shutting_down: bool = False + + +@dataclass +class UnifiedState: + gateway_running: bool = False + gateway_started_at: float = 0.0 + channels: dict[str, ChannelHealthState] = field(default_factory=dict) + config: ConfigHealthState = field(default_factory=ConfigHealthState) + monitoring: MonitoringState = field(default_factory=MonitoringState) + readiness: ReadinessState = field(default_factory=ReadinessState) + diagnostic: DiagnosticState = field(default_factory=DiagnosticState) + connections: GatewayConnectionState = field(default_factory=GatewayConnectionState) + + def to_dict(self) -> dict[str, Any]: + return { + "gateway": { + "running": self.gateway_running, + "started_at": self.gateway_started_at, + }, + "channels": { + key: { + "channel_type": ch.channel_type, + "account_id": ch.account_id, + "state": ch.state, + "connected": ch.connected, + "healthy": ch.healthy, + "health_reason": ch.health_reason, + "health_detail": ch.health_detail, + "last_error": ch.last_error, + "active_runs": ch.active_runs, + "uptime_seconds": ch.uptime_seconds, + "restart_count": ch.restart_count, + } + for key, ch in self.channels.items() + }, + "config": { + "status": self.config.status, + "last_known_good": self.config.last_known_good, + "issues": self.config.issues, + "revision": self.config.revision, + }, + "monitoring": { + "checks_total": self.monitoring.checks_total, + "current_status": self.monitoring.current_status, + "unhealthy_channels": self.monitoring.unhealthy_channels, + "degraded_channels": self.monitoring.degraded_channels, + "last_check_at": self.monitoring.last_check_at, + }, + "readiness": { + "ready": self.readiness.ready, + "failing": self.readiness.failing, + "uptime_ms": self.readiness.uptime_ms, + "event_loop_degraded": self.readiness.event_loop_degraded, + }, + "diagnostic": { + "overall_status": self.diagnostic.overall_status, + "warnings_count": self.diagnostic.warnings_count, + "security_warnings_count": self.diagnostic.security_warnings_count, + }, + "connections": { + "active_connections": self.connections.active_connections, + "shutting_down": self.connections.shutting_down, + }, + } + + @property + def is_healthy(self) -> bool: + return ( + self.readiness.ready + and self.monitoring.current_status == "healthy" + and not self.connections.shutting_down + ) + + +class StateAggregator: + def get_unified_state(self) -> UnifiedState: + state = UnifiedState() + + self._collect_gateway_state(state) + self._collect_channel_health_state(state) + self._collect_config_health_state(state) + self._collect_monitoring_state(state) + self._collect_readiness_state(state) + self._collect_connection_state(state) + + return state + + def _collect_gateway_state(self, state: UnifiedState) -> None: + try: + from yuxi.channel.runtime.manager import gateway + + state.gateway_running = gateway.is_running() + state.gateway_started_at = gateway._started_at + except Exception: + pass + + def _collect_channel_health_state(self, state: UnifiedState) -> None: + try: + from yuxi.channel.monitoring.health_monitor import channel_health_monitor + + metrics = channel_health_monitor.metrics + eval_details = metrics.evaluation_details + + from yuxi.channel.runtime.manager import gateway + + for key, snap in gateway.get_state().channels.items(): + parts = key.split(":", 1) + if len(parts) != 2: + continue + detail = eval_details.get(key, {}) + state.channels[key] = ChannelHealthState( + channel_type=parts[0], + account_id=parts[1], + state=snap.state.value, + connected=snap.connected, + healthy=detail.get("healthy", True), + health_reason=detail.get("reason", "unknown"), + health_detail=detail.get("detail"), + last_event_at=snap.last_event_at, + last_error=snap.last_error, + active_runs=snap.active_runs, + uptime_seconds=snap.uptime_seconds, + restart_count=snap.restart_count, + ) + except Exception: + pass + + def _collect_config_health_state(self, state: UnifiedState) -> None: + try: + pass + except Exception: + pass + + def _collect_monitoring_state(self, state: UnifiedState) -> None: + try: + from yuxi.channel.monitoring.health_monitor import channel_health_monitor + + metrics = channel_health_monitor.metrics + state.monitoring = MonitoringState( + checks_total=metrics.checks_total, + current_status=metrics.current_status, + unhealthy_channels=list(metrics.unhealthy_channels), + degraded_channels=list(metrics.degraded_channels), + last_check_at=metrics.last_check_at, + ) + except Exception: + pass + + def _collect_readiness_state(self, state: UnifiedState) -> None: + try: + from yuxi.channel.monitoring.readiness import readiness_checker + + last = readiness_checker._last_result + if last is not None: + state.readiness = ReadinessState( + ready=last.ready, + failing=list(last.failing), + uptime_ms=last.uptime_ms, + event_loop_degraded=last.event_loop_degraded, + ) + except Exception: + pass + + def _collect_connection_state(self, state: UnifiedState) -> None: + try: + from yuxi.channel.gateway.server import gateway_ws_server + + state.connections = GatewayConnectionState( + active_connections=gateway_ws_server.active_count, + shutting_down=gateway_ws_server.shutting_down, + ) + except Exception: + pass + + +state_aggregator = StateAggregator()