Cheng Zhou
205 lines
6.6 KiB
Python
205 lines
6.6 KiB
Python
"""权限系统监控
|
|
|
|
收集缓存性能指标和告警信息。
|
|
权限检查的统计数据（总次数、允许/拒绝、耗时）已持久化在
|
|
permission_access_logs 表中，通过 /metrics 端点实时聚合查询。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from collections import deque
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
from app.core.permission_cache import get_permission_cache
|
|
|
|
|
|
CACHE_HIT_RATE_HEALTH_MIN_ACCESSES = 50
|
|
CACHE_METRICS_WINDOW_SECONDS = 3600
|
|
|
|
|
|
@dataclass
|
|
class CacheMetrics:
|
|
"""缓存指标"""
|
|
|
|
total_accesses: int = 0
|
|
cache_hits: int = 0
|
|
cache_misses: int = 0
|
|
cache_invalidations: int = 0
|
|
_access_events: deque[tuple[float, str, bool]] = field(
|
|
default_factory=deque,
|
|
repr=False,
|
|
)
|
|
_invalidation_events: deque[float] = field(default_factory=deque, repr=False)
|
|
|
|
@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),
|
|
}
|
|
|
|
def record_access(self, cache_name: str, *, hit: bool) -> None:
|
|
self.total_accesses += 1
|
|
if hit:
|
|
self.cache_hits += 1
|
|
else:
|
|
self.cache_misses += 1
|
|
self._access_events.append((time.time(), cache_name, hit))
|
|
|
|
def record_invalidation(self) -> None:
|
|
self.cache_invalidations += 1
|
|
self._invalidation_events.append(time.time())
|
|
|
|
def window_stats(self, seconds: int = CACHE_METRICS_WINDOW_SECONDS) -> dict[str, Any]:
|
|
cutoff = time.time() - seconds
|
|
while self._access_events and self._access_events[0][0] < cutoff:
|
|
self._access_events.popleft()
|
|
while self._invalidation_events and self._invalidation_events[0] < cutoff:
|
|
self._invalidation_events.popleft()
|
|
|
|
def summarize(cache_name: str | None = None) -> dict[str, Any]:
|
|
events = [
|
|
event for event in self._access_events
|
|
if cache_name is None or event[1] == cache_name
|
|
]
|
|
hits = sum(event[2] for event in events)
|
|
total = len(events)
|
|
misses = total - hits
|
|
return {
|
|
"total_accesses": total,
|
|
"cache_hits": hits,
|
|
"cache_misses": misses,
|
|
"cache_invalidations": len(self._invalidation_events),
|
|
"hit_rate": round(hits / total * 100, 2) if total else 0.0,
|
|
"miss_rate": round(misses / total * 100, 2) if total else 0.0,
|
|
}
|
|
|
|
stats = summarize()
|
|
stats.update({
|
|
"window_seconds": seconds,
|
|
"scope": "sliding_window",
|
|
"by_cache": {
|
|
"project_permissions": summarize("project_permissions"),
|
|
"member_roles": summarize("member_roles"),
|
|
},
|
|
})
|
|
return stats
|
|
|
|
|
|
@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, cache_name: str = "project_permissions") -> None:
|
|
self.metrics.cache_metrics.record_access(cache_name, hit=True)
|
|
|
|
def record_cache_miss(self, cache_name: str = "project_permissions") -> None:
|
|
self.metrics.cache_metrics.record_access(cache_name, hit=False)
|
|
|
|
def record_cache_invalidation(self) -> None:
|
|
self.metrics.cache_metrics.record_invalidation()
|
|
|
|
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.window_stats(),
|
|
"cache_metrics_lifetime": self.metrics.cache_metrics.to_dict(),
|
|
}
|
|
|
|
def get_metrics(self) -> dict[str, Any]:
|
|
return {
|
|
"cache_metrics": self.metrics.cache_metrics.window_stats(),
|
|
"cache_metrics_lifetime": self.metrics.cache_metrics.to_dict(),
|
|
"uptime_seconds": self.metrics.uptime_seconds,
|
|
}
|
|
|
|
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
|