logging_service.py

# -*- coding: utf-8 -*-
"""Location: ./mcpgateway/services/logging_service.py
Copyright contributors to the MCP-CONTEXT-FORGE project
SPDX-License-Identifier: Apache-2.0

Logging Service Implementation.
This module implements structured logging according to the MCP specification.
It supports RFC 5424 severity levels, log level management, and log event subscriptions.
"""

# Standard
import asyncio
from asyncio.events import AbstractEventLoop
from datetime import datetime, timezone
import logging
from logging.handlers import RotatingFileHandler
import os
import socket
from typing import Any, AsyncGenerator, Dict, List, NotRequired, Optional, TextIO, TypedDict

# Third-Party
from pythonjsonlogger import json as jsonlogger  # You may need to install python-json-logger package

# First-Party
from mcpgateway.common.models import LogLevel
from mcpgateway.config import settings
from mcpgateway.services.log_storage_service import LogStorageService
from mcpgateway.utils.correlation_id import get_correlation_id
from mcpgateway.utils.url_auth import sanitize_exception_message

# Optional OpenTelemetry support (Third-Party)
try:
    # Third-Party
    from opentelemetry import trace  # type: ignore[import-untyped]
except ImportError:
    trace = None  # type: ignore[assignment]

AnyioClosedResourceError: Optional[type]  # pylint: disable=invalid-name
try:
    # Optional import; only used for filtering a known benign upstream error (Third-Party)
    # Third-Party
    from anyio import ClosedResourceError as AnyioClosedResourceError  # pylint: disable=invalid-name
except Exception:  # pragma: no cover - environment without anyio
    AnyioClosedResourceError = None  # pylint: disable=invalid-name

# First-Party
# Standard log format used across the codebase
LOG_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
LOG_DATE_FORMAT = "%Y-%m-%dT%H:%M:%S"

# Cache static values at module load - these don't change during process lifetime
_CACHED_HOSTNAME: str = socket.gethostname()
_CACHED_PID: int = os.getpid()

# Cache level mapping dictionaries at module load to avoid recreation on every log call
# Maps Python log level names to MCP LogLevel enum (used in StorageHandler.emit)
_PYTHON_TO_MCP_LEVEL_MAP: Dict[str, LogLevel] = {
    "DEBUG": LogLevel.DEBUG,
    "INFO": LogLevel.INFO,
    "WARNING": LogLevel.WARNING,
    "ERROR": LogLevel.ERROR,
    "CRITICAL": LogLevel.CRITICAL,
}

# Maps MCP LogLevel to Python logging method names (used in notify)
_MCP_TO_PYTHON_METHOD_MAP: Dict[LogLevel, str] = {
    LogLevel.DEBUG: "debug",
    LogLevel.INFO: "info",
    LogLevel.NOTICE: "info",  # Map NOTICE to INFO
    LogLevel.WARNING: "warning",
    LogLevel.ERROR: "error",
    LogLevel.CRITICAL: "critical",
    LogLevel.ALERT: "critical",  # Map ALERT to CRITICAL
    LogLevel.EMERGENCY: "critical",  # Map EMERGENCY to CRITICAL
}

# Maps MCP LogLevel to numeric values for comparison (used in _should_log)
_MCP_LEVEL_VALUES: Dict[LogLevel, int] = {
    LogLevel.DEBUG: 0,
    LogLevel.INFO: 1,
    LogLevel.NOTICE: 2,
    LogLevel.WARNING: 3,
    LogLevel.ERROR: 4,
    LogLevel.CRITICAL: 5,
    LogLevel.ALERT: 6,
    LogLevel.EMERGENCY: 7,
}

# Create a text formatter with standard format
text_formatter = logging.Formatter(LOG_FORMAT, datefmt=LOG_DATE_FORMAT)


class CorrelationIdJsonFormatter(jsonlogger.JsonFormatter):
    """JSON formatter that includes correlation ID and OpenTelemetry trace context."""

    def add_fields(self, log_record: dict, record: logging.LogRecord, message_dict: dict) -> None:  # pylint: disable=arguments-renamed
        """Add custom fields to the log record.

        Args:
            log_record: The dictionary that will be logged as JSON
            record: The original LogRecord
            message_dict: Additional message fields

        """
        super().add_fields(log_record, record, message_dict)

        # Add timestamp in ISO 8601 format with 'Z' suffix for UTC
        dt = datetime.fromtimestamp(record.created, tz=timezone.utc)
        log_record["@timestamp"] = dt.isoformat().replace("+00:00", "Z")

        # Add hostname and process ID for log aggregation - use cached values for performance
        log_record["hostname"] = _CACHED_HOSTNAME
        log_record["process_id"] = _CACHED_PID

        # Add correlation ID from context
        correlation_id = get_correlation_id()
        if correlation_id:
            log_record["request_id"] = correlation_id

        # Add OpenTelemetry trace context if available
        if trace is not None:
            try:
                span = trace.get_current_span()
                if span and span.is_recording():
                    span_context = span.get_span_context()
                    if span_context.is_valid:
                        # Format trace_id and span_id as hex strings
                        log_record["trace_id"] = format(span_context.trace_id, "032x")
                        log_record["span_id"] = format(span_context.span_id, "016x")
                        log_record["trace_flags"] = format(span_context.trace_flags, "02x")
            except Exception:  # nosec B110 - intentionally catching all exceptions for optional tracing
                # Error accessing span context, continue without trace fields
                pass


# Create a JSON formatter with correlation ID support (uses same base format)
json_formatter = CorrelationIdJsonFormatter(LOG_FORMAT, datefmt=LOG_DATE_FORMAT)

# Note: Don't use basicConfig here as it conflicts with our custom dual logging setup
# The LoggingService.initialize() method will properly configure all handlers

# Global handlers will be created lazily
_file_handler: Optional[logging.Handler] = None
_text_handler: Optional[logging.StreamHandler[TextIO]] = None


def _get_file_handler() -> logging.Handler:
    """Get or create the file handler.

    Returns:
        logging.Handler: Either a RotatingFileHandler or regular FileHandler for JSON logging.

    Raises:
        ValueError: If file logging is disabled or no log file specified.

    """
    global _file_handler  # pylint: disable=global-statement
    if _file_handler is None:
        # Only create if file logging is enabled and file is specified
        if not settings.log_to_file or not settings.log_file:
            raise ValueError("File logging is disabled or no log file specified")

        # Ensure log folder exists
        if settings.log_folder:
            os.makedirs(settings.log_folder, exist_ok=True)
            log_path = os.path.join(settings.log_folder, settings.log_file)
        else:
            log_path = settings.log_file

        # Create appropriate handler based on rotation settings
        if settings.log_rotation_enabled:
            max_bytes = settings.log_max_size_mb * 1024 * 1024  # Convert MB to bytes
            _file_handler = RotatingFileHandler(log_path, maxBytes=max_bytes, backupCount=settings.log_backup_count, mode=settings.log_filemode)
        else:
            _file_handler = logging.FileHandler(log_path, mode=settings.log_filemode)

        _file_handler.setFormatter(json_formatter)
    return _file_handler


def _get_text_handler() -> logging.StreamHandler[TextIO]:
    """Get or create the text handler.

    Returns:
        logging.StreamHandler: The stream handler for console logging.

    """
    global _text_handler  # pylint: disable=global-statement
    if _text_handler is None:
        _text_handler = logging.StreamHandler()
        _text_handler.setFormatter(text_formatter)
    return _text_handler


class StorageHandler(logging.Handler):
    """Custom logging handler that stores logs in LogStorageService."""

    def __init__(self, storage_service: LogStorageService):
        """Initialize the storage handler.

        Args:
            storage_service: The LogStorageService instance to store logs in

        """
        super().__init__()
        self.storage = storage_service
        self.loop: AbstractEventLoop | None = None

    def emit(self, record: logging.LogRecord) -> None:
        """Emit a log record to storage.

        Args:
            record: The LogRecord to emit

        """
        if not self.storage:
            return

        # Map Python log levels to MCP LogLevel (uses module-level cached dict)
        log_level = _PYTHON_TO_MCP_LEVEL_MAP.get(record.levelname, LogLevel.INFO)

        # Extract entity context from record if available
        entity_type = getattr(record, "entity_type", None)
        entity_id = getattr(record, "entity_id", None)
        entity_name = getattr(record, "entity_name", None)
        request_id = getattr(record, "request_id", None)

        # Format the message
        try:
            message = self.format(record)
        except Exception:
            message = record.getMessage()

        # Store the log asynchronously
        try:
            coro = self.storage.add_log(
                level=log_level,
                message=message,
                entity_type=entity_type,
                entity_id=entity_id,
                entity_name=entity_name,
                logger=record.name,
                request_id=request_id,
            )

            try:
                # Fast path: we're already on an event loop thread.
                loop = asyncio.get_running_loop()
                self.loop = loop
                task = loop.create_task(coro)
                task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
            except RuntimeError:
                # Fallback: no running loop in this thread; attempt to schedule on a known loop.
                loop = self.loop
                if loop is None or not loop.is_running():
                    coro.close()
                    return

                future = asyncio.run_coroutine_threadsafe(coro, loop)
                future.add_done_callback(lambda f: f.exception() if not f.cancelled() else None)
        except Exception:
            # Silently fail to avoid logging recursion
            pass  # nosec B110 - Intentional to prevent logging recursion


class _LogMessageData(TypedDict):
    """Log message data structure."""

    level: LogLevel
    data: Any
    timestamp: str
    logger: NotRequired[str]


class _LogMessage(TypedDict):
    """Log message event structure."""

    type: str
    data: _LogMessageData


class LoggingService:
    """MCP logging service.

    Implements structured logging with:
    - RFC 5424 severity levels
    - Log level management
    - Log event subscriptions
    - Logger name tracking
    """

    def __init__(self) -> None:
        """Initialize logging service."""
        self._level = LogLevel.INFO
        self._subscribers: List[asyncio.Queue[_LogMessage]] = []
        self._loggers: Dict[str, logging.Logger] = {}
        self._storage: LogStorageService | None = None  # Will be initialized if admin UI is enabled
        self._storage_handler: Optional[StorageHandler] = None  # Track the storage handler for cleanup

    async def initialize(self) -> None:
        """Initialize logging service.

        Examples:
            >>> from mcpgateway.services.logging_service import LoggingService
            >>> import asyncio
            >>> service = LoggingService()
            >>> asyncio.run(service.initialize())

        """
        # Update service log level from settings BEFORE configuring loggers
        self._level = LogLevel[settings.log_level.upper()]

        root_logger = logging.getLogger()
        self._loggers[""] = root_logger

        # Clear existing handlers to avoid duplicates
        root_logger.handlers.clear()

        # Set root logger level to match settings - this is critical for LOG_LEVEL to work
        log_level = getattr(logging, settings.log_level.upper())
        root_logger.setLevel(log_level)

        # Console handler (stdout/stderr)
        #
        # LOG_FORMAT controls the console output format:
        # - text: human-friendly
        # - json: machine-friendly (Loki/ELK) and includes OTEL trace context when available
        if getattr(settings, "log_format", "text").lower() == "json":
            console_handler = logging.StreamHandler()
            console_handler.setFormatter(json_formatter)
        else:
            console_handler = _get_text_handler()
        console_handler.setLevel(log_level)
        root_logger.addHandler(console_handler)

        # Only add file handler if enabled
        if settings.log_to_file and settings.log_file:
            try:
                file_handler = _get_file_handler()
                file_handler.setLevel(log_level)
                root_logger.addHandler(file_handler)
                if settings.log_rotation_enabled:
                    logging.info(
                        "File logging enabled with rotation: %s/%s (max: %sMB, backups: %s)", settings.log_folder or ".", settings.log_file, settings.log_max_size_mb, settings.log_backup_count
                    )
                else:
                    logging.info("File logging enabled (no rotation): %s/%s", settings.log_folder or ".", settings.log_file)
            except Exception as e:
                logging.warning("Failed to initialize file logging: %s", e)
        else:
            logging.info("File logging disabled - logging to stdout/stderr only")

        # Configure uvicorn loggers to use our handlers (for access logs)
        # Note: This needs to be done both at init and dynamically as uvicorn creates loggers later
        self._configure_uvicorn_loggers()

        # Initialize log storage if admin UI is enabled
        if settings.mcpgateway_ui_enabled or settings.mcpgateway_admin_api_enabled:
            self._storage = LogStorageService()

            # Add storage handler to capture all logs
            self._storage_handler = StorageHandler(self._storage)
            self._storage_handler.setFormatter(text_formatter)
            self._storage_handler.setLevel(log_level)
            root_logger.addHandler(self._storage_handler)

            logging.info("Log storage initialized with %sMB buffer", settings.log_buffer_size_mb)

        logging.info("Logging service initialized")

        # Suppress noisy upstream logs for normal stream closures in MCP streamable HTTP
        self._install_closedresourceerror_filter()

        # Redact sensitive query parameters from httpx/httpcore log messages
        self._install_httpx_url_sanitize_filter()

        # Suppress high-volume health check and readiness probe logs
        self._install_uvicorn_health_check_filter()

    async def shutdown(self) -> None:
        """Shutdown logging service.

        Examples:
            >>> from mcpgateway.services.logging_service import LoggingService
            >>> import asyncio
            >>> service = LoggingService()
            >>> asyncio.run(service.shutdown())

        """
        # Remove storage handler from root logger if it was added
        if self._storage_handler:
            root_logger = logging.getLogger()
            root_logger.removeHandler(self._storage_handler)
            self._storage_handler = None

        # Clear subscribers
        self._subscribers.clear()
        logging.info("Logging service shutdown")

    def _install_closedresourceerror_filter(self) -> None:
        """Install a filter to drop benign ClosedResourceError logs from upstream MCP.

        The MCP streamable HTTP server logs an ERROR when the in-memory channel is
        closed during normal client disconnects, raising ``anyio.ClosedResourceError``.
        This filter suppresses those specific records to keep logs clean.

        Examples:
            >>> # Initialize service (installs filter)
            >>> import asyncio, logging, anyio
            >>> service = LoggingService()
            >>> asyncio.run(service.initialize())
            >>> # Locate the installed filter on the target logger
            >>> target = logging.getLogger('mcp.server.streamable_http')
            >>> flts = [f for f in target.filters if f.__class__.__name__.endswith('SuppressClosedResourceErrorFilter')]
            >>> len(flts) >= 1
            True
            >>> filt = flts[0]
            >>> # Non-target logger should pass through even if message matches
            >>> rec_other = logging.makeLogRecord({'name': 'other.logger', 'msg': 'ClosedResourceError'})
            >>> filt.filter(rec_other)
            True
            >>> # Target logger with message containing ClosedResourceError should be suppressed
            >>> rec_target_msg = logging.makeLogRecord({'name': 'mcp.server.streamable_http', 'msg': 'ClosedResourceError in normal shutdown'})
            >>> filt.filter(rec_target_msg)
            False
            >>> # Target logger with ClosedResourceError in exc_info should be suppressed
            >>> try:
            ...     raise anyio.ClosedResourceError
            ... except anyio.ClosedResourceError as e:
            ...     rec_target_exc = logging.makeLogRecord({
            ...         'name': 'mcp.server.streamable_http',
            ...         'msg': 'Error in message router',
            ...         'exc_info': (e.__class__, e, None),
            ...     })
            >>> filt.filter(rec_target_exc)
            False
            >>> # Cleanup
            >>> asyncio.run(service.shutdown())

        """

        class _SuppressClosedResourceErrorFilter(logging.Filter):
            """Filter to suppress ClosedResourceError exceptions from MCP streamable HTTP logger.

            This filter prevents noisy ClosedResourceError exceptions from the upstream
            MCP streamable HTTP implementation from cluttering the logs. These errors
            are typically harmless connection cleanup events.
            """

            def filter(self, record: logging.LogRecord) -> bool:  # noqa: D401
                """Filter log records to suppress ClosedResourceError exceptions.

                Args:
                    record: The log record to evaluate

                Returns:
                    True to allow the record through, False to suppress it

                """
                # Apply only to upstream MCP streamable HTTP logger
                if not record.name.startswith("mcp.server.streamable_http"):
                    return True

                # If exception info is present, check its type
                exc_info = getattr(record, "exc_info", None)
                if exc_info and AnyioClosedResourceError is not None:
                    exc_type, exc, _tb = exc_info
                    try:
                        if isinstance(exc, AnyioClosedResourceError) or (getattr(exc_type, "__name__", "") == "ClosedResourceError"):
                            return False
                    except Exception:
                        # Be permissive if anything goes wrong, don't drop logs accidentally
                        return True

                # Fallback: drop if message text clearly indicates ClosedResourceError
                try:
                    msg = record.getMessage()
                    if "ClosedResourceError" in msg:
                        return False
                except Exception:
                    pass  # nosec B110 - Intentional to prevent logging recursion
                return True

        target_logger = logging.getLogger("mcp.server.streamable_http")
        target_logger.addFilter(_SuppressClosedResourceErrorFilter())

    @staticmethod
    def _install_uvicorn_health_check_filter() -> None:
        """Install a filter to suppress health check and readiness probe logs from uvicorn.access.

        Kubernetes and other orchestrators frequently poll /health and /ready endpoints,
        generating high-volume, low-value access logs. This filter suppresses those logs
        to reduce noise while preserving logs for other endpoints.

        The filter checks the formatted message string for /health and /ready patterns.

        Examples:
            >>> import asyncio, logging
            >>> service = LoggingService()
            >>> asyncio.run(service.initialize())
            >>> filt = [f for f in logging.getLogger('uvicorn.access').filters
            ...         if f.__class__.__name__ == '_UvicornHealthCheckFilter'][0]
            >>> rec = logging.LogRecord(
            ...     name='uvicorn.access', level=logging.INFO, pathname='', lineno=0,
            ...     msg='10.254.16.2:53664 - "GET /ready HTTP/1.1" 200',
            ...     args=(),
            ...     exc_info=None
            ... )
            >>> filt.filter(rec)
            False
            >>> rec2 = logging.LogRecord(
            ...     name='uvicorn.access', level=logging.INFO, pathname='', lineno=0,
            ...     msg='10.254.16.2:53664 - "GET /health HTTP/1.1" 200',
            ...     args=(),
            ...     exc_info=None
            ... )
            >>> filt.filter(rec2)
            False
            >>> rec3 = logging.LogRecord(
            ...     name='uvicorn.access', level=logging.INFO, pathname='', lineno=0,
            ...     msg='10.254.20.2:34444 - "GET /api/tools HTTP/1.1" 200',
            ...     args=(),
            ...     exc_info=None
            ... )
            >>> filt.filter(rec3)
            True
            >>> asyncio.run(service.shutdown())

        """

        class _UvicornHealthCheckFilter(logging.Filter):
            """Filter to suppress health check and readiness probe logs from uvicorn.access.

            Checks the formatted log message for /health and /ready endpoint patterns.
            """

            def filter(self, record: logging.LogRecord) -> bool:  # noqa: D401
                """Filter log records to suppress health/ready endpoint access logs.

                Args:
                    record: The log record to evaluate.

                Returns:
                    True to allow the record through, False to suppress it

                """
                # Only apply to uvicorn.access logger
                if not record.name.startswith("uvicorn.access"):
                    return True

                try:
                    # Check the formatted message for health/ready endpoints
                    if hasattr(record, "getMessage"):
                        msg = record.getMessage()
                        # Match patterns like: 'GET /health HTTP' or '"GET /ready HTTP"'
                        if "GET /health" in msg or "GET /ready" in msg:
                            return False
                except Exception:
                    pass  # nosec B110 - Never break logging due to filter failure
                return True

        uvicorn_access_logger = logging.getLogger("uvicorn.access")
        uvicorn_access_logger.addFilter(_UvicornHealthCheckFilter())

    @staticmethod
    def _install_httpx_url_sanitize_filter() -> None:
        """Install a filter to redact sensitive query parameters from httpx/httpcore log messages.

        httpx and httpcore log full request URLs at INFO level, bypassing
        application-level sanitization. This filter intercepts those log
        records and redacts sensitive query parameters (api_key, token, etc.)
        before they reach any handler.

        Examples:
            >>> import asyncio, logging
            >>> service = LoggingService()
            >>> asyncio.run(service.initialize())
            >>> filt = [f for f in logging.getLogger('httpx').filters
            ...         if f.__class__.__name__ == '_HttpxUrlSanitizeFilter'][0]
            >>> rec = logging.makeLogRecord({
            ...     'name': 'httpx',
            ...     'msg': 'HTTP Request: GET https://example.mcp.server.com/sse?api_key=secret-value "HTTP/1.1 200 OK"',
            ... })
            >>> filt.filter(rec)
            True
            >>> 'secret-value' not in rec.getMessage()
            True
            >>> 'api_key=REDACTED' in rec.getMessage()
            True
            >>> asyncio.run(service.shutdown())

        """

        class _HttpxUrlSanitizeFilter(logging.Filter):
            """Filter that redacts sensitive query parameters from URLs in httpx log messages."""

            def filter(self, record: logging.LogRecord) -> bool:  # noqa: D401
                """Sanitize URLs in the log record message, then allow it through.

                Args:
                    record: The log record to sanitize.

                Returns:
                    Always True (record is never suppressed, only sanitized).

                """
                try:
                    msg = record.getMessage()
                    sanitized = sanitize_exception_message(msg)
                    if sanitized != msg:
                        record.msg = sanitized
                        record.args = None
                except Exception:
                    pass  # nosec B110 - Never break logging due to sanitization failure
                return True

        url_filter = _HttpxUrlSanitizeFilter()
        for logger_name in ("httpx", "httpcore"):
            logging.getLogger(logger_name).addFilter(url_filter)

    def get_logger(self, name: str) -> logging.Logger:
        """Get or create logger instance.

        Args:
            name: Logger name

        Returns:
            Logger instance

        Examples:
            >>> from mcpgateway.services.logging_service import LoggingService
            >>> service = LoggingService()
            >>> logger = service.get_logger('test')
            >>> import logging
            >>> isinstance(logger, logging.Logger)
            True

        """
        if name not in self._loggers:
            logger = logging.getLogger(name)

            # Don't add handlers to child loggers - let them inherit from root
            # This prevents duplicate logging while maintaining dual output (console + file)
            logger.propagate = True

            # Don't set level on child loggers - let them inherit from root logger
            # This ensures LOG_LEVEL environment variable is respected after initialize() runs
            # The root logger level is set in initialize() based on settings.log_level

            self._loggers[name] = logger

        return self._loggers[name]

    async def set_level(self, level: LogLevel) -> None:
        """Set minimum log level.

        This updates the level for all registered loggers.

        Args:
            level: New log level

        Examples:
            >>> from mcpgateway.services.logging_service import LoggingService
            >>> from mcpgateway.common.models import LogLevel
            >>> import asyncio
            >>> service = LoggingService()
            >>> asyncio.run(service.set_level(LogLevel.DEBUG))

        """
        self._level = level

        # Update all loggers and handlers
        log_level = getattr(logging, level.upper())

        # Update Python root logger so new child loggers inherit the correct level
        logging.getLogger().setLevel(log_level)

        # Update handler levels so they don't filter out records the logger passes
        for handler in logging.getLogger().handlers:
            handler.setLevel(log_level)

        for logger in self._loggers.values():
            logger.setLevel(log_level)

        await self.notify(f"Log level set to {level}", LogLevel.INFO, "logging")

    async def notify(  # pylint: disable=too-many-positional-arguments
        self,
        data: Any,
        level: LogLevel,
        logger_name: Optional[str] = None,
        entity_type: Optional[str] = None,
        entity_id: Optional[str] = None,
        entity_name: Optional[str] = None,
        request_id: Optional[str] = None,
        extra_data: Optional[Dict[str, Any]] = None,
    ) -> None:
        """Send log notification to subscribers.

        Args:
            data: Log message data
            level: Log severity level
            logger_name: Optional logger name
            entity_type: Type of entity (tool, resource, server, gateway)
            entity_id: ID of the related entity
            entity_name: Name of the related entity
            request_id: Associated request ID for tracing
            extra_data: Additional structured data

        Examples:
            >>> from mcpgateway.services.logging_service import LoggingService
            >>> from mcpgateway.common.models import LogLevel
            >>> import asyncio
            >>> service = LoggingService()
            >>> asyncio.run(service.notify('test', LogLevel.INFO))

        """
        # Skip if below current level
        if not self._should_log(level):
            return

        # Format notification message
        message: _LogMessage = {
            "type": "log",
            "data": {
                "level": level,
                "data": data,
                "timestamp": datetime.now(timezone.utc).isoformat(),
            },
        }
        if logger_name:
            message["data"]["logger"] = logger_name

        # Log through standard logging
        logger = self.get_logger(logger_name or "")

        # Map MCP log levels to Python logging levels (uses module-level cached dict)
        log_method = _MCP_TO_PYTHON_METHOD_MAP.get(level, "info")
        log_func = getattr(logger, log_method)
        log_func(data)

        # Store in log storage if available
        if self._storage:
            await self._storage.add_log(
                level=level,
                message=str(data),
                entity_type=entity_type,
                entity_id=entity_id,
                entity_name=entity_name,
                logger=logger_name,
                data=extra_data,
                request_id=request_id,
            )

        # Notify subscribers
        for queue in self._subscribers:
            try:
                await queue.put(message)
            except Exception as e:
                logger.error("Failed to notify subscriber: %s", e)

    async def subscribe(self) -> AsyncGenerator[_LogMessage, None]:
        """Subscribe to log messages.

        Returns a generator yielding log message events.

        Yields:
            Log message events

        Examples:
            This example was removed to prevent the test runner from hanging on async generator consumption.

        """
        queue: asyncio.Queue[_LogMessage] = asyncio.Queue()
        self._subscribers.append(queue)
        try:
            while True:
                message = await queue.get()
                yield message
        finally:
            self._subscribers.remove(queue)

    def _should_log(self, level: LogLevel) -> bool:
        """Check if level meets minimum threshold.

        Args:
            level: Log level to check

        Returns:
            True if should log

        Examples:
            >>> from mcpgateway.common.models import LogLevel
            >>> service = LoggingService()
            >>> service._level = LogLevel.WARNING
            >>> service._should_log(LogLevel.ERROR)
            True
            >>> service._should_log(LogLevel.INFO)
            False
            >>> service._should_log(LogLevel.WARNING)
            True
            >>> service._should_log(LogLevel.DEBUG)
            False

        """
        # Uses module-level cached dict for performance
        return _MCP_LEVEL_VALUES[level] >= _MCP_LEVEL_VALUES[self._level]

    def _configure_uvicorn_loggers(self) -> None:
        """Configure uvicorn loggers to use our dual logging setup.

        This method handles uvicorn's logging setup which can happen after our initialization.
        Uvicorn creates its own loggers and handlers, so we need to redirect them to our setup.
        """
        uvicorn_loggers = ["uvicorn", "uvicorn.access", "uvicorn.error", "uvicorn.asgi"]

        for logger_name in uvicorn_loggers:
            uvicorn_logger = logging.getLogger(logger_name)

            # Clear any handlers that uvicorn may have added
            uvicorn_logger.handlers.clear()

            # Make sure they propagate to root (which has our dual handlers)
            uvicorn_logger.propagate = True

            # Set level to match our logging service level
            if hasattr(self, "_level"):
                log_level = getattr(logging, self._level.upper())
                uvicorn_logger.setLevel(log_level)

            # Track the logger
            self._loggers[logger_name] = uvicorn_logger

    def configure_uvicorn_after_startup(self) -> None:
        """Public method to reconfigure uvicorn loggers after server startup.

        Call this after uvicorn has started to ensure access logs go to dual output.
        This handles the case where uvicorn creates loggers after our initialization.
        """
        self._configure_uvicorn_loggers()
        logging.info("Uvicorn loggers reconfigured for dual logging")

    def get_storage(self) -> Optional[LogStorageService]:
        """Get the log storage service if available.

        Returns:
            LogStorageService instance or None if not initialized

        """
        return self._storage