Cheng Zhou
145 lines
4.2 KiB
Python
145 lines
4.2 KiB
Python
"""权限系统监控
|
|
|
|
收集缓存性能指标和告警信息。
|
|
权限检查的统计数据（总次数、允许/拒绝、耗时）已持久化在
|
|
permission_access_logs 表中，通过 /metrics 端点实时聚合查询。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
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 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:
|
|
cache_metrics: CacheMetrics = field(default_factory=CacheMetrics)
|
|
start_time: float = field(default_factory=time.time)
|
|
|
|
@property
|
|
def uptime_seconds(self) -> float:
|
|
return time.time() - self.start_time
|
|
|
|
def reset(self) -> None:
|
|
self.cache_metrics = CacheMetrics()
|
|
self.start_time = time.time()
|
|
|
|
|
|
class PermissionMonitor:
|
|
"""权限系统监控器（仅缓存指标 + 告警）"""
|
|
|
|
def __init__(self):
|
|
self.metrics = PermissionSystemMetrics()
|
|
self._alerts: list[dict[str, Any]] = []
|
|
|
|
def record_cache_hit(self) -> None:
|
|
self.metrics.cache_metrics.total_accesses += 1
|
|
self.metrics.cache_metrics.cache_hits += 1
|
|
|
|
def record_cache_miss(self) -> None:
|
|
self.metrics.cache_metrics.total_accesses += 1
|
|
self.metrics.cache_metrics.cache_misses += 1
|
|
|
|
def record_cache_invalidation(self) -> None:
|
|
self.metrics.cache_metrics.cache_invalidations += 1
|
|
|
|
def record_error_alert(self, error: Exception) -> None:
|
|
self._add_alert(
|
|
level="error",
|
|
type="permission_check_error",
|
|
message=f"权限检查出错: {str(error)}",
|
|
data={"error": str(error)},
|
|
)
|
|
|
|
def record_slow_check_alert(self, elapsed_ms: float) -> None:
|
|
if elapsed_ms > 50:
|
|
self._add_alert(
|
|
level="warning",
|
|
type="slow_permission_check",
|
|
message=f"权限检查耗时过长: {elapsed_ms:.2f}ms",
|
|
data={"elapsed_ms": elapsed_ms},
|
|
)
|
|
|
|
def _add_alert(self, level: str, type: str, message: str, data: dict[str, Any] | None = None) -> None:
|
|
self._alerts.append({
|
|
"timestamp": time.time(),
|
|
"level": level,
|
|
"type": type,
|
|
"message": message,
|
|
"data": data or {},
|
|
})
|
|
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]]:
|
|
alerts = self._alerts
|
|
if level:
|
|
alerts = [a for a in alerts if a["level"] == level]
|
|
return alerts[-limit:]
|
|
|
|
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
|