feat: suppress uvicorn access logs for /health and /ready probes

docs/docs/manage/logging.md

@@ -31,6 +31,9 @@ ContextForge uses dual-format logging:
 }
 ```
 
+!!! note "Health Check Log Suppression"
+    Access logs for `/health` and `/ready` endpoints are automatically suppressed to reduce noise from Kubernetes health probes and readiness checks. This filtering happens at the logging layer and cannot be disabled. Other endpoints are logged normally.
+
 ### Text Format (Console Output)
 ```
 2025-01-09 17:30:15,123 - mcpgateway.gateway_service - INFO - Registered gateway: peer-gateway-1

mcpgateway/services/logging_service.py

@@ -374,6 +374,9 @@ async def initialize(self) -> None:
         # 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.
 
@@ -481,6 +484,84 @@ def filter(self, record: logging.LogRecord) -> bool:  # noqa: D401
         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.

tests/unit/mcpgateway/services/test_logging_service.py

@@ -227,6 +227,122 @@ async def test_subscribe_cleanup_removes_queue_on_cancel():
     assert len(service._subscribers) == 0
 
 
+# ---------------------------------------------------------------------------
+# uvicorn health check suppression filter
+# ---------------------------------------------------------------------------
+
+
+class TestUvicornHealthCheckFilter:
+    """Tests for the uvicorn.access health check suppression filter."""
+
+    @staticmethod
+    def _get_filter():
+        """Install the filter and return it."""
+        LoggingService._install_uvicorn_health_check_filter()
+        uvicorn_logger = logging.getLogger("uvicorn.access")
+        return [f for f in uvicorn_logger.filters if f.__class__.__name__ == "_UvicornHealthCheckFilter"][-1]
+
+    def test_suppresses_ready_endpoint(self):
+        """Test that /ready endpoint logs are suppressed."""
+        filt = self._get_filter()
+        record = logging.LogRecord(
+            name="uvicorn.access",
+            level=logging.INFO,
+            pathname="",
+            lineno=0,
+            msg='10.254.20.2:34442 - "GET /ready HTTP/1.1" 200',
+            args=(),
+            exc_info=None,
+        )
+        result = filt.filter(record)
+        assert result is False
+
+    def test_suppresses_health_endpoint(self):
+        """Test that /health endpoint logs are suppressed."""
+        filt = self._get_filter()
+        record = logging.LogRecord(
+            name="uvicorn.access",
+            level=logging.INFO,
+            pathname="",
+            lineno=0,
+            msg='10.254.20.2:34444 - "GET /health HTTP/1.1" 200',
+            args=(),
+            exc_info=None,
+        )
+        result = filt.filter(record)
+        assert result is False
+
+    def test_allows_other_endpoints(self):
+        """Test that other endpoints are not suppressed."""
+        filt = self._get_filter()
+        record = logging.LogRecord(
+            name="uvicorn.access",
+            level=logging.INFO,
+            pathname="",
+            lineno=0,
+            msg='10.254.20.2:34446 - "GET /api/tools HTTP/1.1" 200',
+            args=(),
+            exc_info=None,
+        )
+        result = filt.filter(record)
+        assert result is True
+
+    def test_allows_non_uvicorn_access_loggers(self):
+        """Test that non-uvicorn.access loggers are not affected."""
+        filt = self._get_filter()
+        record = logging.LogRecord(
+            name="uvicorn.error",
+            level=logging.INFO,
+            pathname="",
+            lineno=0,
+            msg='10.254.20.2:34442 - "GET /ready HTTP/1.1" 200',
+            args=(),
+            exc_info=None,
+        )
+        result = filt.filter(record)
+        assert result is True
+
+    def test_handles_getmessage_exception(self):
+        """Test that filter handles getMessage() exceptions gracefully."""
+        filt = self._get_filter()
+        record = logging.LogRecord(
+            name="uvicorn.access",
+            level=logging.INFO,
+            pathname="",
+            lineno=0,
+            msg="test",
+            args=(),
+            exc_info=None,
+        )
+        # Simulate getMessage raising an exception
+        with patch.object(record, "getMessage", side_effect=RuntimeError("mocked error")):
+            result = filt.filter(record)
+        # Should return True (allow the log) rather than crashing
+        assert result is True
+
+    def test_handles_missing_args_gracefully(self):
+        """Test that filter handles records without health/ready patterns."""
+        filt = self._get_filter()
+        record = logging.LogRecord(name="uvicorn.access", level=logging.INFO, pathname="", lineno=0, msg="Some message without health or ready", args=(), exc_info=None)
+        result = filt.filter(record)
+        assert result is True
+
+    def test_handles_malformed_messages_gracefully(self):
+        """Test that filter handles malformed messages gracefully."""
+        filt = self._get_filter()
+        record = logging.LogRecord(name="uvicorn.access", level=logging.INFO, pathname="", lineno=0, msg="GET /api/users", args=(), exc_info=None)
+        result = filt.filter(record)
+        assert result is True
+
+    def test_filter_survives_exception(self):
+        """Test that filter survives exceptions during processing."""
+        filt = self._get_filter()
+        record = logging.LogRecord(name="uvicorn.access", level=logging.INFO, pathname="", lineno=0, msg="test", args=None, exc_info=None)
+        # Should not raise even with None args
+        result = filt.filter(record)
+        assert result is True
+
+
 # ---------------------------------------------------------------------------
 # httpx URL sanitize filter
 # ---------------------------------------------------------------------------