Spaces:

jun2024
/

fc

Paused

fc / apps /core /infrastructure /monitoring.py

jun2024

Upload 2241 files

b4f2238 verified 7 months ago

8.33 kB

"""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,
                },
            )

	"""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,
	},
	)