"""Module for monitoring.""" from __future__ import annotations """ 性能监控模块 提供 API 响应时间监控和数据库查询次数监控 """ import logging import os import time from collections.abc import Callable from contextlib import contextmanager from functools import wraps from typing import Any from django.conf import settings from django.db import connection, reset_queries logger = logging.getLogger("apps.core.monitoring") class PerformanceMonitor: """ 性能监控器 监控 API 响应时间和数据库查询次数 """ # 性能阈值配置 SLOW_API_THRESHOLD_MS = 1000 # API 响应时间阈值(毫秒) SLOW_QUERY_THRESHOLD_MS = 100 # 慢查询阈值(毫秒) MAX_QUERY_COUNT = 10 # 最大查询次数阈值 @classmethod def _should_collect_queries(cls) -> bool: env = (os.environ.get("DJANGO_DB_QUERY_METRICS", "") or "").lower().strip() enabled = env in ("true", "1", "yes") return bool(getattr(settings, "DEBUG", False) or enabled) @classmethod def monitor_api(cls, endpoint: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]: """ API 性能监控装饰器 监控 API 响应时间和数据库查询次数 Args: endpoint: API 端点名称 Usage: @monitor_api("create_case") def create_case(request, data) -> None: ... """ def decorator(func: Callable[..., Any]) -> Callable[..., Any]: @wraps(func) def wrapper(*args: Any, **kwargs: Any) -> Any: collect_queries = cls._should_collect_queries() prev_force_debug_cursor = getattr(connection, "force_debug_cursor", False) if collect_queries and not prev_force_debug_cursor: connection.force_debug_cursor = True if collect_queries: reset_queries() # 记录开始时间 start_time = time.time() start_query_count = len(connection.queries) if collect_queries else 0 try: # 执行函数 result = func(*args, **kwargs) # 计算性能指标 duration_ms = (time.time() - start_time) * 1000 query_count = len(connection.queries) - start_query_count if collect_queries else 0 # 记录性能日志 cls._log_performance( endpoint=endpoint, duration_ms=duration_ms, query_count=query_count, query_count_collected=collect_queries, success=True, ) # 检查性能问题 cls._check_performance_issues(endpoint=endpoint, duration_ms=duration_ms, query_count=query_count) return result except Exception as e: # 记录失败的性能日志 duration_ms = (time.time() - start_time) * 1000 query_count = len(connection.queries) - start_query_count if collect_queries else 0 cls._log_performance( endpoint=endpoint, duration_ms=duration_ms, query_count=query_count, query_count_collected=collect_queries, success=False, error=str(e), ) raise finally: if collect_queries and not prev_force_debug_cursor: connection.force_debug_cursor = prev_force_debug_cursor return wrapper return decorator @classmethod @contextmanager def monitor_operation(cls, operation_name: str) -> Any: """ 操作性能监控上下文管理器 监控任意操作的性能 Args: operation_name: 操作名称 Usage: with monitor_operation("fetch_external_data"): data = fetch_data() """ collect_queries = cls._should_collect_queries() prev_force_debug_cursor = getattr(connection, "force_debug_cursor", False) if collect_queries and not prev_force_debug_cursor: connection.force_debug_cursor = True if collect_queries: reset_queries() # 记录开始时间 start_time = time.time() start_query_count = len(connection.queries) if collect_queries else 0 try: yield # 计算性能指标 duration_ms = (time.time() - start_time) * 1000 query_count = len(connection.queries) - start_query_count if collect_queries else 0 # 记录性能日志 cls._log_performance( endpoint=operation_name, duration_ms=duration_ms, query_count=query_count, query_count_collected=collect_queries, success=True, ) # 检查性能问题 cls._check_performance_issues(endpoint=operation_name, duration_ms=duration_ms, query_count=query_count) except Exception as e: # 记录失败的性能日志 duration_ms = (time.time() - start_time) * 1000 query_count = len(connection.queries) - start_query_count if collect_queries else 0 cls._log_performance( endpoint=operation_name, duration_ms=duration_ms, query_count=query_count, query_count_collected=collect_queries, success=False, error=str(e), ) raise finally: if collect_queries and not prev_force_debug_cursor: connection.force_debug_cursor = prev_force_debug_cursor @classmethod def _log_performance( cls, endpoint: str, duration_ms: float, query_count: int, query_count_collected: bool, success: bool, error: str | None = None, ) -> None: """ 记录性能日志 Args: endpoint: 端点名称 duration_ms: 响应时间(毫秒) query_count: 查询次数 success: 是否成功 error: 错误信息(可选) """ log_data = { "endpoint": endpoint, "duration_ms": round(duration_ms, 2), "query_count": query_count, "query_count_collected": bool(query_count_collected), "success": success, "metric_type": "performance", } if error: log_data["error"] = error # 根据性能情况选择日志级别 if not success: logger.error(f"API 执行失败: {endpoint}", extra=log_data) elif duration_ms > cls.SLOW_API_THRESHOLD_MS: logger.warning(f"慢 API 检测: {endpoint}", extra=log_data) else: logger.info(f"API 执行完成: {endpoint}", extra=log_data) @classmethod def _check_performance_issues(cls, endpoint: str, duration_ms: float, query_count: int) -> None: """ 检查性能问题 Args: endpoint: 端点名称 duration_ms: 响应时间(毫秒) query_count: 查询次数 """ issues = [] # 检查响应时间 if duration_ms > cls.SLOW_API_THRESHOLD_MS: issues.append(f"响应时间过长: {duration_ms:.2f}ms (阈值: {cls.SLOW_API_THRESHOLD_MS}ms)") # 检查查询次数 if query_count > cls.MAX_QUERY_COUNT: issues.append(f"查询次数过多: {query_count} 次 (阈值: {cls.MAX_QUERY_COUNT} 次,可能存在 N+1 查询问题)") # 记录性能问题 if issues: logger.warning( f"性能问题检测: {endpoint}", extra={ "endpoint": endpoint, "issues": issues, "duration_ms": round(duration_ms, 2), "query_count": query_count, }, )