monitoring.py

"""
Monitoring and Health Checks
监控和健康检查系统，包括指标收集、日志聚合、告警等
"""

import asyncio
import logging
import json
import time
from typing import Dict, Any, List, Optional, Callable
from datetime import datetime, timedelta
from dataclasses import dataclass, asdict, field
from collections import defaultdict, deque
import threading
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import psutil

# Optional imports for FastAPI integration
try:
    from fastapi import FastAPI, HTTPException
    FASTAPI_AVAILABLE = True
except ImportError:
    FASTAPI_AVAILABLE = False

logger = logging.getLogger(__name__)


@dataclass
class MetricPoint:
    """指标数据点"""
    timestamp: float
    value: float
    labels: Dict[str, str] = field(default_factory=dict)


@dataclass
class ServiceHealth:
    """服务健康状态"""
    service_name: str
    status: str  # "healthy", "unhealthy", "degraded", "unknown"
    last_check: datetime
    response_time: float
    error_rate: float
    availability: float
    details: Dict[str, Any] = field(default_factory=dict)


class MetricsCollector:
    """统一指标收集器"""

    def __init__(self, service_name: str, buffer_size: int = 1000):
        self.service_name = service_name
        self.buffer_size = buffer_size
        self.metrics_buffer = defaultdict(lambda: deque(maxlen=buffer_size))

        # Prometheus指标 - 使用try-except避免重复注册
        try:
            self.request_counter = Counter(
                f'{service_name}_requests_total',
                'Total number of requests',
                ['method', 'status']
            )
        except:
            # 如果指标已存在，获取已存在的指标
            from prometheus_client import REGISTRY
            self.request_counter = REGISTRY._names_to_collectors.get(f'{service_name}_requests_total')

        try:
            self.request_duration = Histogram(
                f'{service_name}_request_duration_seconds',
                'Request duration in seconds',
                ['method']
            )
        except:
            from prometheus_client import REGISTRY
            self.request_duration = REGISTRY._names_to_collectors.get(f'{service_name}_request_duration_seconds')

        try:
            self.error_counter = Counter(
                f'{service_name}_errors_total',
                'Total number of errors',
                ['error_type']
            )
        except:
            from prometheus_client import REGISTRY
            self.error_counter = REGISTRY._names_to_collectors.get(f'{service_name}_errors_total')

        try:
            self.active_requests = Gauge(
                f'{service_name}_active_requests',
                'Number of active requests'
            )
        except:
            from prometheus_client import REGISTRY
            self.active_requests = REGISTRY._names_to_collectors.get(f'{service_name}_active_requests')

        try:
            self.queue_depth = Gauge(
                f'{service_name}_queue_depth',
                'Queue depth',
                ['queue_name']
            )
        except:
            from prometheus_client import REGISTRY
            self.queue_depth = REGISTRY._names_to_collectors.get(f'{service_name}_queue_depth')

    def record_request(self, method: str, status: str, duration: float):
        """记录请求指标"""
        self.request_counter.labels(method=method, status=status).inc()
        self.request_duration.labels(method=method).observe(duration)

        # 记录到缓冲区
        metric_point = MetricPoint(
            timestamp=time.time(),
            value=duration,
            labels={"method": method, "status": status}
        )
        self.metrics_buffer[f"{method}_duration"].append(metric_point)

    def record_error(self, error_type: str):
        """记录错误"""
        self.error_counter.labels(error_type=error_type).inc()

    def set_active_requests(self, count: int):
        """设置活跃请求数"""
        self.active_requests.set(count)

    def set_queue_depth(self, queue_name: str, depth: int):
        """设置队列深度"""
        self.queue_depth.labels(queue_name=queue_name).set(depth)

    def get_metrics_summary(self, metric_name: str, duration_seconds: int = 300) -> Dict[str, Any]:
        """获取指标摘要"""
        current_time = time.time()
        cutoff_time = current_time - duration_seconds

        if metric_name not in self.metrics_buffer:
            return {}

        recent_points = [
            point for point in self.metrics_buffer[metric_name]
            if point.timestamp >= cutoff_time
        ]

        if not recent_points:
            return {}

        values = [point.value for point in recent_points]
        return {
            "count": len(values),
            "avg": sum(values) / len(values),
            "min": min(values),
            "max": max(values),
            "p50": sorted(values)[len(values) // 2],
            "p95": sorted(values)[int(len(values) * 0.95)],
            "p99": sorted(values)[int(len(values) * 0.99)]
        }


class HealthChecker:
    """健康检查器"""

    def __init__(self, check_interval: int = 30):
        self.check_interval = check_interval
        self.health_checks = {}
        self.health_history = defaultdict(lambda: deque(maxlen=100))
        self._running = False

    def register_health_check(self, service_name: str, check_func: Callable):
        """注册健康检查函数"""
        self.health_checks[service_name] = check_func

    async def perform_health_check(self, service_name: str) -> ServiceHealth:
        """执行健康检查"""
        if service_name not in self.health_checks:
            return ServiceHealth(
                service_name=service_name,
                status="unknown",
                last_check=datetime.utcnow(),
                response_time=0,
                error_rate=0,
                availability=0,
                details={"error": "No health check registered"}
            )

        start_time = time.time()
        try:
            # 执行健康检查
            is_healthy = await self.health_checks[service_name]()
            response_time = time.time() - start_time

            # 计算历史指标
            history = self.health_history[service_name]
            history.append(is_healthy)

            if len(history) > 0:
                availability = sum(history) / len(history) * 100
                error_rate = (len(history) - sum(history)) / len(history) * 100
            else:
                availability = 100 if is_healthy else 0
                error_rate = 0 if is_healthy else 100

            # 确定状态
            if is_healthy:
                if availability >= 95:
                    status = "healthy"
                elif availability >= 80:
                    status = "degraded"
                else:
                    status = "unhealthy"
            else:
                status = "unhealthy"

            health = ServiceHealth(
                service_name=service_name,
                status=status,
                last_check=datetime.utcnow(),
                response_time=response_time,
                error_rate=error_rate,
                availability=availability
            )

            self.health_history[service_name].append(health)
            return health

        except Exception as e:
            response_time = time.time() - start_time
            logger.error(f"Health check failed for {service_name}: {e}")

            return ServiceHealth(
                service_name=service_name,
                status="unhealthy",
                last_check=datetime.utcnow(),
                response_time=response_time,
                error_rate=100,
                availability=0,
                details={"error": str(e)}
            )

    async def check_all_services(self) -> Dict[str, ServiceHealth]:
        """检查所有服务"""
        results = {}
        for service_name in self.health_checks:
            results[service_name] = await self.perform_health_check(service_name)
        return results


class AlertManager:
    """告警管理器"""

    def __init__(self, alert_thresholds: Dict[str, float] = None):
        self.alert_thresholds = alert_thresholds or {
            "error_rate": 10.0,      # 错误率超过10%
            "response_time": 5.0,    # 响应时间超过5秒
            "availability": 95.0,    # 可用性低于95%
            "queue_depth": 1000      # 队列深度超过1000
        }
        self.active_alerts = {}
        self.alert_history = deque(maxlen=1000)

    def check_alerts(self, metrics: Dict[str, Any], health_status: Dict[str, ServiceHealth]):
        """检查告警条件"""
        current_time = datetime.utcnow()

        # 检查错误率告警
        if "error_rate" in metrics and metrics["error_rate"] > self.alert_thresholds["error_rate"]:
            self._trigger_alert("high_error_rate", f"Error rate {metrics['error_rate']:.1f}% exceeds threshold", {
                "error_rate": metrics["error_rate"],
                "threshold": self.alert_thresholds["error_rate"]
            })

        # 检查响应时间告警
        if "avg_response_time" in metrics and metrics["avg_response_time"] > self.alert_thresholds["response_time"]:
            self._trigger_alert("high_response_time", f"Average response time {metrics['avg_response_time']:.2f}s exceeds threshold", {
                "response_time": metrics["avg_response_time"],
                "threshold": self.alert_thresholds["response_time"]
            })

        # 检查可用性告警
        for service_name, health in health_status.items():
            if health.availability < self.alert_thresholds["availability"]:
                self._trigger_alert("low_availability", f"Service {service_name} availability {health.availability:.1f}% below threshold", {
                    "service": service_name,
                    "availability": health.availability,
                    "threshold": self.alert_thresholds["availability"]
                })

    def _trigger_alert(self, alert_type: str, message: str, details: Dict[str, Any]):
        """触发告警"""
        alert_key = f"{alert_type}_{details.get('service', 'system')}"

        if alert_key not in self.active_alerts:
            alert = {
                "type": alert_type,
                "message": message,
                "details": details,
                "triggered_at": datetime.utcnow(),
                "severity": self._get_alert_severity(alert_type),
                "acknowledged": False
            }
            self.active_alerts[alert_key] = alert
            self.alert_history.append(alert)

            logger.warning(f"ALERT TRIGGERED: {message}")

    def _get_alert_severity(self, alert_type: str) -> str:
        """获取告警严重程度"""
        severity_map = {
            "high_error_rate": "critical",
            "high_response_time": "warning",
            "low_availability": "critical",
            "high_queue_depth": "warning"
        }
        return severity_map.get(alert_type, "info")

    def acknowledge_alert(self, alert_key: str):
        """确认告警"""
        if alert_key in self.active_alerts:
            self.active_alerts[alert_key]["acknowledged"] = True
            self.active_alerts[alert_key]["acknowledged_at"] = datetime.utcnow()

    def get_active_alerts(self) -> Dict[str, Dict]:
        """获取活跃告警"""
        return self.active_alerts.copy()

    def get_alert_history(self, limit: int = 100) -> List[Dict]:
        """获取告警历史"""
        return list(self.alert_history)[-limit:]


class SystemMonitor:
    """系统监控器"""

    def __init__(self, service_name: str):
        self.service_name = service_name
        self.metrics_collector = MetricsCollector(service_name)
        self.health_checker = HealthChecker()
        self.alert_manager = AlertManager()
        self._monitoring_task = None

    async def start_monitoring(self):
        """开始监控"""
        # 启动Prometheus指标服务器
        try:
            start_http_server(9090)  # Prometheus默认端口
            logger.info("Prometheus metrics server started on port 9090")
        except Exception as e:
            logger.warning(f"Failed to start Prometheus server: {e}")

        # 启动后台监控任务
        self._monitoring_task = asyncio.create_task(self._monitoring_loop())

    async def stop_monitoring(self):
        """停止监控"""
        if self._monitoring_task:
            self._monitoring_task.cancel()
            try:
                await self._monitoring_task
            except asyncio.CancelledError:
                pass

    async def _monitoring_loop(self):
        """监控循环"""
        while True:
            try:
                # 收集系统指标
                system_metrics = self._collect_system_metrics()

                # 检查所有服务健康状态
                health_status = await self.health_checker.check_all_services()

                # 检查告警
                self.alert_manager.check_alerts(system_metrics, health_status)

                # 记录指标
                logger.info(f"System metrics: {json.dumps(system_metrics, indent=2)}")
                
                # 转换 health_status 为可序列化的格式
                health_status_serializable = {}
                for k, v in health_status.items():
                    v_dict = asdict(v)
                    # 转换 datetime 为 ISO 格式字符串
                    if 'last_check' in v_dict and v_dict['last_check']:
                        v_dict['last_check'] = v_dict['last_check'].isoformat() if hasattr(v_dict['last_check'], 'isoformat') else str(v_dict['last_check'])
                    health_status_serializable[k] = v_dict
                
                logger.info(f"Health status: {json.dumps(health_status_serializable, indent=2)}")

                await asyncio.sleep(30)  # 每30秒检查一次

            except Exception as e:
                logger.error(f"Monitoring loop error: {e}")
                await asyncio.sleep(60)  # 出错时等待更长时间

    def _collect_system_metrics(self) -> Dict[str, Any]:
        """收集系统指标"""
        try:
            # CPU使用率
            cpu_percent = psutil.cpu_percent(interval=1)

            # 内存使用
            memory = psutil.virtual_memory()
            memory_percent = memory.percent
            memory_available = memory.available / (1024 ** 3)  # GB

            # 磁盘使用
            disk = psutil.disk_usage('/')
            disk_percent = (disk.used / disk.total) * 100

            # 网络IO
            net_io = psutil.net_io_counters()
            network_bytes_sent = net_io.bytes_sent
            network_bytes_recv = net_io.bytes_recv

            return {
                "timestamp": datetime.utcnow().isoformat(),
                "cpu_percent": cpu_percent,
                "memory_percent": memory_percent,
                "memory_available_gb": memory_available,
                "disk_percent": disk_percent,
                "network_bytes_sent": network_bytes_sent,
                "network_bytes_recv": network_bytes_recv
            }

        except Exception as e:
            logger.error(f"Failed to collect system metrics: {e}")
            return {}

    def register_health_check(self, service_name: str, check_func: Callable):
        """注册健康检查"""
        self.health_checker.register_health_check(service_name, check_func)

    def record_request(self, method: str, status: str, duration: float):
        """记录请求"""
        self.metrics_collector.record_request(method, status, duration)

    def get_dashboard_data(self) -> Dict[str, Any]:
        """获取仪表板数据"""
        return {
            "service_name": self.service_name,
            "timestamp": datetime.utcnow().isoformat(),
            "system_metrics": self._collect_system_metrics(),
            "service_metrics": self.metrics_collector.get_metrics_summary("all", 3600),
            "health_status": {
                service: asdict(health)
                for service, health in self.health_checker.check_all_services()
            },
            "active_alerts": self.alert_manager.get_active_alerts(),
            "recent_alerts": self.alert_manager.get_alert_history(10)
        }


# 监控API端点
def create_monitoring_app(monitor: SystemMonitor):
    """创建监控API应用"""
    if not FASTAPI_AVAILABLE:
        logger.warning("FastAPI not available, skipping monitoring app creation")
        return None

    app = FastAPI(title="System Monitoring API", version="1.0.0")

    @app.get("/health")
    async def health_check():
        """健康检查端点"""
        return monitor.health_checker.check_all_services()

    @app.get("/metrics")
    async def get_metrics():
        """获取指标"""
        return monitor.metrics_collector.get_metrics_summary("all", 3600)

    @app.get("/dashboard")
    async def get_dashboard():
        """获取仪表板数据"""
        return monitor.get_dashboard_data()

    @app.get("/alerts")
    async def get_alerts():
        """获取告警"""
        return {
            "active_alerts": monitor.alert_manager.get_active_alerts(),
            "alert_history": monitor.alert_manager.get_alert_history(20)
        }

    @app.post("/alerts/acknowledge/{alert_key}")
    async def acknowledge_alert(alert_key: str):
        """确认告警"""
        monitor.alert_manager.acknowledge_alert(alert_key)
        return {"status": "acknowledged"}

    return app


# 使用示例
async def example_monitoring():
    """监控使用示例"""
    # 创建系统监控器
    monitor = SystemMonitor("code_assistant_system")

    # 注册健康检查
    async def mock_health_check() -> bool:
        # 模拟健康检查
        import random
        return random.random() > 0.1  # 90%的健康率

    monitor.register_health_check("coder_service", mock_health_check)
    monitor.register_health_check("checker_service", mock_health_check)
    monitor.register_health_check("debugger_service", mock_health_check)

    # 启动监控
    await monitor.start_monitoring()

    # 模拟一些请求
    for i in range(10):
        monitor.record_request("generate", "success" if i < 8 else "failed", 0.5 + i * 0.1)
        await asyncio.sleep(1)

    # 获取仪表板数据
    dashboard_data = monitor.get_dashboard_data()
    print(f"Dashboard data: {json.dumps(dashboard_data, indent=2, default=str)}")

    # 停止监控
    await monitor.stop_monitoring()


if __name__ == "__main__":
    asyncio.run(example_monitoring())