import threading import time from collections import defaultdict, deque from dataclasses import dataclass from datetime import datetime, timedelta from typing import Dict, List, Any import psutil from fastapi import Request, Response from fastapi.responses import PlainTextResponse from prometheus_client import Counter, Histogram, Gauge, generate_latest, CONTENT_TYPE_LATEST from prometheus_fastapi_instrumentator import Instrumentator from app.core.config import settings from app.log import logger @dataclass class RequestMetrics: """ 请求指标数据类 """ path: str method: str status_code: int response_time: float timestamp: datetime client_ip: str user_agent: str @dataclass class PerformanceSnapshot: """ 性能快照数据类 """ timestamp: datetime cpu_usage: float memory_usage: float active_requests: int request_rate: float avg_response_time: float error_rate: float slow_requests: int class FastAPIMonitor: """ FastAPI性能监控器 """ def __init__(self, max_history: int = 1000, window_size: int = 60): self.max_history = max_history self.window_size = window_size # 秒 # 请求历史记录 self.request_history: deque = deque(maxlen=max_history) # 实时统计 self.active_requests = 0 self.total_requests = 0 self.error_requests = 0 self.slow_requests = 0 # 响应时间超过1秒的请求 # 时间窗口统计 self.window_requests: deque = deque(maxlen=window_size) self.window_response_times: deque = deque(maxlen=window_size) # 线程锁 self._lock = threading.Lock() # 性能阈值 self.slow_request_threshold = 1.0 # 1秒 self.error_threshold = 0.05 # 5% self.cpu_threshold = 80.0 # 80% self.memory_threshold = 80.0 # 80% # 告警状态 self.alerts: List[str] = [] logger.info("FastAPI性能监控器已初始化") def record_request(self, request: Request, response: Response, response_time: float): """ 记录请求指标 """ with self._lock: # 创建请求指标 metrics = RequestMetrics( path=str(request.url.path), method=request.method, status_code=response.status_code, response_time=response_time, timestamp=datetime.now(), client_ip=request.client.host if request.client else "unknown", user_agent=request.headers.get("user-agent", "unknown") ) # 添加到历史记录 self.request_history.append(metrics) # 更新统计 self.total_requests += 1 if response.status_code >= 400: self.error_requests += 1 if response_time > self.slow_request_threshold: self.slow_requests += 1 # 添加到时间窗口 self.window_requests.append(metrics) self.window_response_times.append(response_time) def start_request(self): """ 开始处理请求 """ with self._lock: self.active_requests += 1 def end_request(self): """ 结束处理请求 """ with self._lock: self.active_requests = max(0, self.active_requests - 1) def get_performance_snapshot(self) -> PerformanceSnapshot: """ 获取性能快照 """ with self._lock: now = datetime.now() # 计算请求率（每分钟） recent_requests = [ req for req in self.window_requests if now - req.timestamp < timedelta(seconds=self.window_size) ] request_rate = len(recent_requests) / (self.window_size / 60) # 计算平均响应时间 recent_response_times = [ rt for rt in self.window_response_times if len(self.window_response_times) > 0 ] avg_response_time = sum(recent_response_times) / len(recent_response_times) if recent_response_times else 0 # 计算错误率 error_rate = self.error_requests / self.total_requests if self.total_requests > 0 else 0 # 系统资源使用率 cpu_usage = psutil.cpu_percent(interval=0.1) memory_usage = psutil.virtual_memory().percent return PerformanceSnapshot( timestamp=now, cpu_usage=cpu_usage, memory_usage=memory_usage, active_requests=self.active_requests, request_rate=request_rate, avg_response_time=avg_response_time, error_rate=error_rate, slow_requests=self.slow_requests ) def get_top_endpoints(self, limit: int = 10) -> List[Dict[str, Any]]: """ 获取最活跃的端点 """ with self._lock: endpoint_stats = defaultdict(lambda: { 'count': 0, 'total_time': 0, 'errors': 0, 'avg_time': 0 }) for req in self.request_history: key = f"{req.method} {req.path}" endpoint_stats[key]['count'] += 1 endpoint_stats[key]['total_time'] += req.response_time if req.status_code >= 400: endpoint_stats[key]['errors'] += 1 # 计算平均时间 for stats in endpoint_stats.values(): if stats['count'] > 0: stats['avg_time'] = stats['total_time'] / stats['count'] # 按请求数量排序 sorted_endpoints = sorted( [{'endpoint': k, **v} for k, v in endpoint_stats.items()], key=lambda x: x['count'], reverse=True ) return sorted_endpoints[:limit] def get_recent_errors(self, limit: int = 20) -> List[Dict[str, Any]]: """ 获取最近的错误请求 """ with self._lock: errors = [ { 'timestamp': req.timestamp.isoformat(), 'method': req.method, 'path': req.path, 'status_code': req.status_code, 'response_time': req.response_time, 'client_ip': req.client_ip } for req in self.request_history if req.status_code >= 400 ] return errors[-limit:] def check_alerts(self) -> List[str]: """ 检查告警条件 """ snapshot = self.get_performance_snapshot() alerts = [] if snapshot.error_rate > self.error_threshold: alerts.append(f"错误率过高: {snapshot.error_rate:.2%}") if snapshot.cpu_usage > self.cpu_threshold: alerts.append(f"CPU使用率过高: {snapshot.cpu_usage:.1f}%") if snapshot.memory_usage > self.memory_threshold: alerts.append(f"内存使用率过高: {snapshot.memory_usage:.1f}%") if snapshot.avg_response_time > self.slow_request_threshold: alerts.append(f"平均响应时间过长: {snapshot.avg_response_time:.2f}s") if snapshot.request_rate > 1000: # 每分钟1000请求 alerts.append(f"请求率过高: {snapshot.request_rate:.0f} req/min") self.alerts = alerts return alerts # 全局监控实例 monitor = FastAPIMonitor() def setup_prometheus_metrics(app): """ 设置Prometheus指标 """ if not settings.PERFORMANCE_MONITOR_ENABLE: return # 创建Prometheus指标 request_counter = Counter( "http_requests_total", "Total number of HTTP requests", ["method", "endpoint", "status"] ) request_duration = Histogram( "http_request_duration_seconds", "HTTP request duration in seconds", ["method", "endpoint"] ) active_requests = Gauge( "http_active_requests", "Number of active HTTP requests" ) # 自定义指标收集函数 def custom_metrics(request: Request, response: Response, response_time: float): request_counter.labels( method=request.method, endpoint=request.url.path, status=response.status_code ).inc() request_duration.labels( method=request.method, endpoint=request.url.path ).observe(response_time) active_requests.set(monitor.active_requests) # 设置Prometheus监控 Instrumentator().instrument(app).expose(app, include_in_schema=False, should_gzip=True) # 添加自定义指标 @app.middleware("http") async def monitor_middleware(request: Request, call_next): start_time = time.time() # 开始请求 monitor.start_request() try: response = await call_next(request) response_time = time.time() - start_time # 记录请求指标 monitor.record_request(request, response, response_time) # 更新Prometheus指标 custom_metrics(request, response, response_time) return response except Exception as e: response_time = time.time() - start_time logger.error(f"请求处理异常: {e}") # 创建错误响应 response = Response( content=str(e), status_code=500, media_type="text/plain" ) # 记录错误请求 monitor.record_request(request, response, response_time) return response finally: # 结束请求 monitor.end_request() logger.info("Prometheus指标监控已设置") def get_metrics_response(): """ 获取Prometheus指标响应 """ return PlainTextResponse( generate_latest(), media_type=CONTENT_TYPE_LATEST )