ctms/backend/app/core/permission_monitor.py

"""权限系统监控

实现权限系统的监控功能，包括：
- 权限检查统计
- 缓存性能监控
- 异常检测
- 性能指标收集
"""

from __future__ import annotations

import time
import uuid
from dataclasses import dataclass, field
from typing import Any

from app.core.permission_cache import get_permission_cache


@dataclass
class PermissionCheckMetrics:
    """权限检查指标"""

    total_checks: int = 0  # 总检查次数
    allowed_checks: int = 0  # 允许的检查次数
    denied_checks: int = 0  # 拒绝的检查次数
    total_time: float = 0.0  # 总耗时（秒）
    min_time: float = float("inf")  # 最小耗时（秒）
    max_time: float = 0.0  # 最大耗时（秒）
    errors: int = 0  # 错误次数

    @property
    def avg_time(self) -> float:
        """平均耗时（秒）"""
        if self.total_checks == 0:
            return 0.0
        return self.total_time / self.total_checks

    @property
    def allow_rate(self) -> float:
        """允许率（百分比）"""
        if self.total_checks == 0:
            return 0.0
        return (self.allowed_checks / self.total_checks) * 100

    @property
    def deny_rate(self) -> float:
        """拒绝率（百分比）"""
        if self.total_checks == 0:
            return 0.0
        return (self.denied_checks / self.total_checks) * 100

    @property
    def error_rate(self) -> float:
        """错误率（百分比）"""
        if self.total_checks == 0:
            return 0.0
        return (self.errors / self.total_checks) * 100

    def to_dict(self) -> dict[str, Any]:
        """转换为字典"""
        return {
            "total_checks": self.total_checks,
            "allowed_checks": self.allowed_checks,
            "denied_checks": self.denied_checks,
            "total_time": round(self.total_time, 3),
            "min_time": round(self.min_time, 3) if self.min_time != float("inf") else 0,
            "max_time": round(self.max_time, 3),
            "avg_time": round(self.avg_time, 3),
            "allow_rate": round(self.allow_rate, 2),
            "deny_rate": round(self.deny_rate, 2),
            "error_rate": round(self.error_rate, 2),
            "errors": self.errors,
        }


@dataclass
class CacheMetrics:
    """缓存指标"""

    total_accesses: int = 0  # 总访问次数
    cache_hits: int = 0  # 缓存命中次数
    cache_misses: int = 0  # 缓存未命中次数
    cache_invalidations: int = 0  # 缓存失效次数

    @property
    def hit_rate(self) -> float:
        """缓存命中率（百分比）"""
        if self.total_accesses == 0:
            return 0.0
        return (self.cache_hits / self.total_accesses) * 100

    @property
    def miss_rate(self) -> float:
        """缓存未命中率（百分比）"""
        if self.total_accesses == 0:
            return 0.0
        return (self.cache_misses / self.total_accesses) * 100

    def to_dict(self) -> dict[str, Any]:
        """转换为字典"""
        return {
            "total_accesses": self.total_accesses,
            "cache_hits": self.cache_hits,
            "cache_misses": self.cache_misses,
            "cache_invalidations": self.cache_invalidations,
            "hit_rate": round(self.hit_rate, 2),
            "miss_rate": round(self.miss_rate, 2),
        }


@dataclass
class PermissionSystemMetrics:
    """权限系统指标"""

    check_metrics: PermissionCheckMetrics = field(default_factory=PermissionCheckMetrics)
    cache_metrics: CacheMetrics = field(default_factory=CacheMetrics)
    last_reset_time: float = field(default_factory=time.time)

    def reset(self) -> None:
        """重置所有指标"""
        self.check_metrics = PermissionCheckMetrics()
        self.cache_metrics = CacheMetrics()
        self.last_reset_time = time.time()

    def to_dict(self) -> dict[str, Any]:
        """转换为字典"""
        return {
            "check_metrics": self.check_metrics.to_dict(),
            "cache_metrics": self.cache_metrics.to_dict(),
            "uptime_seconds": time.time() - self.last_reset_time,
        }


class PermissionMonitor:
    """权限系统监控器

    收集权限系统的运行指标，用于监控和告警。
    """

    def __init__(self):
        """初始化监控器"""
        self.metrics = PermissionSystemMetrics()
        self._alerts: list[dict[str, Any]] = []

    def record_permission_check(
        self,
        allowed: bool,
        elapsed_time: float,
        error: Exception | None = None,
    ) -> None:
        """记录权限检查

        Args:
            allowed: 是否允许
            elapsed_time: 耗时（秒）
            error: 错误对象（如果有）
        """
        metrics = self.metrics.check_metrics
        metrics.total_checks += 1

        if allowed:
            metrics.allowed_checks += 1
        else:
            metrics.denied_checks += 1

        metrics.total_time += elapsed_time
        metrics.min_time = min(metrics.min_time, elapsed_time)
        metrics.max_time = max(metrics.max_time, elapsed_time)

        if error:
            metrics.errors += 1
            self._check_error_alert(error)

        # 检查性能告警
        self._check_performance_alert(elapsed_time)

    def record_cache_hit(self) -> None:
        """记录缓存命中"""
        metrics = self.metrics.cache_metrics
        metrics.total_accesses += 1
        metrics.cache_hits += 1

    def record_cache_miss(self) -> None:
        """记录缓存未命中"""
        metrics = self.metrics.cache_metrics
        metrics.total_accesses += 1
        metrics.cache_misses += 1

    def record_cache_invalidation(self) -> None:
        """记录缓存失效"""
        self.metrics.cache_metrics.cache_invalidations += 1

    def _check_performance_alert(self, elapsed_time: float) -> None:
        """检查性能告警

        如果权限检查耗时过长，发出告警。
        """
        if elapsed_time > 0.05:  # 50ms
            self._add_alert(
                level="warning",
                type="slow_permission_check",
                message=f"权限检查耗时过长: {elapsed_time*1000:.2f}ms",
                data={"elapsed_time": elapsed_time},
            )

    def _check_error_alert(self, error: Exception) -> None:
        """检查错误告警"""
        self._add_alert(
            level="error",
            type="permission_check_error",
            message=f"权限检查出错: {str(error)}",
            data={"error": str(error)},
        )

    def _add_alert(
        self,
        level: str,
        type: str,
        message: str,
        data: dict[str, Any] | None = None,
    ) -> None:
        """添加告警

        Args:
            level: 告警级别 (info, warning, error)
            type: 告警类型
            message: 告警消息
            data: 额外数据
        """
        alert = {
            "timestamp": time.time(),
            "level": level,
            "type": type,
            "message": message,
            "data": data or {},
        }
        self._alerts.append(alert)

        # 只保留最近1000条告警
        if len(self._alerts) > 1000:
            self._alerts = self._alerts[-1000:]

    def get_alerts(self, level: str | None = None, limit: int = 100) -> list[dict[str, Any]]:
        """获取告警列表

        Args:
            level: 告警级别过滤（可选）
            limit: 返回的最大告警数

        Returns:
            告警列表
        """
        alerts = self._alerts
        if level:
            alerts = [a for a in alerts if a["level"] == level]
        return alerts[-limit:]

    def get_metrics(self) -> dict[str, Any]:
        """获取指标"""
        return self.metrics.to_dict()

    def get_cache_stats(self) -> dict[str, Any]:
        """获取缓存统计"""
        cache = get_permission_cache()
        return {
            "cache_items": cache.get_cache_stats(),
            "cache_metrics": self.metrics.cache_metrics.to_dict(),
        }

    def reset_metrics(self) -> None:
        """重置指标"""
        self.metrics.reset()

    def clear_alerts(self) -> None:
        """清除所有告警"""
        self._alerts.clear()


# 全局监控器实例
_permission_monitor: PermissionMonitor | None = None


def get_permission_monitor() -> PermissionMonitor:
    """获取全局权限监控器实例"""
    global _permission_monitor
    if _permission_monitor is None:
        _permission_monitor = PermissionMonitor()
    return _permission_monitor


def set_permission_monitor(monitor: PermissionMonitor) -> None:
    """设置全局权限监控器实例（用于测试）"""
    global _permission_monitor
    _permission_monitor = monitor