Cheng Zhou
335 lines
9.7 KiB
Python
335 lines
9.7 KiB
Python
"""权限系统监控
|
|
|
|
实现权限系统的监控功能，包括：
|
|
- 权限检查统计
|
|
- 缓存性能监控
|
|
- 异常检测
|
|
- 性能指标收集
|
|
"""
|
|
|
|
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
|
|
|
|
|
|
CACHE_HIT_RATE_HEALTH_MIN_ACCESSES = 10
|
|
|
|
|
|
@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
|
|
|
|
|
|
def evaluate_permission_system_health(metrics: dict[str, Any], cache_stats: dict[str, Any]) -> dict[str, Any]:
|
|
"""根据权限系统指标评估健康状态"""
|
|
check_metrics = metrics["check_metrics"]
|
|
cache_metrics = metrics["cache_metrics"]
|
|
|
|
health_score = 100
|
|
issues = []
|
|
|
|
if check_metrics["error_rate"] > 1:
|
|
health_score -= 20
|
|
issues.append("权限检查错误率过高")
|
|
|
|
if (
|
|
cache_metrics["total_accesses"] >= CACHE_HIT_RATE_HEALTH_MIN_ACCESSES
|
|
and cache_metrics["hit_rate"] < 50
|
|
):
|
|
health_score -= 10
|
|
issues.append("缓存命中率过低")
|
|
|
|
if check_metrics["avg_time"] > 0.01:
|
|
health_score -= 10
|
|
issues.append("权限检查响应时间过长")
|
|
|
|
if check_metrics["deny_rate"] > 50:
|
|
health_score -= 5
|
|
issues.append("权限拒绝率过高")
|
|
|
|
return {
|
|
"status": "healthy" if health_score >= 80 else "degraded" if health_score >= 50 else "unhealthy",
|
|
"health_score": max(0, health_score),
|
|
"issues": issues,
|
|
"metrics": metrics,
|
|
"cache_stats": cache_stats,
|
|
}
|