fix(serving): bound validation error logging

nemo_gym/server_utils.py

@@ -15,6 +15,7 @@
 import asyncio
 import atexit
 import json
+import logging
 import resource
 import socket
 import sys
@@ -80,9 +81,58 @@
 from nemo_gym.telemetry.span_groups import GymSpanGroup
 
 
+logger = logging.getLogger(__name__)
+
 _GLOBAL_AIOHTTP_CLIENT: Union[None, ClientSession] = None
 _GLOBAL_AIOHTTP_CLIENT_REQUEST_DEBUG: bool = False
 _UPSTREAM_ERROR_LOG_BODY_CHARS = 2000
+_VALIDATION_ERROR_LOG_BODY_CHARS = 4096
+_VALIDATION_ERROR_LOG_MAX_ERRORS = 20
+
+
+async def _log_validation_exception(request: Request, exc: RequestValidationError) -> None:
+    errors = exc.errors()
+    error_summaries = [
+        {
+            "type": error.get("type"),
+            "loc": error.get("loc"),
+            "msg": error.get("msg"),
+        }
+        for error in errors[:_VALIDATION_ERROR_LOG_MAX_ERRORS]
+    ]
+
+    try:
+        body = await request.body()
+    except Exception:
+        logger.warning(
+            "Request validation failed; request body unavailable",
+            extra={
+                "validation_error_count": len(errors),
+                "validation_errors": error_summaries,
+                "validation_errors_truncated": len(errors) > len(error_summaries),
+            },
+        )
+        return
+
+    raw_prefix = body[:_VALIDATION_ERROR_LOG_BODY_CHARS]
+    rendered_prefix = json.dumps(raw_prefix.decode("utf-8", errors="replace"), ensure_ascii=True)
+    escaped_prefix = rendered_prefix[:_VALIDATION_ERROR_LOG_BODY_CHARS]
+    logger.warning(
+        "Request validation failed",
+        extra={
+            "request_body_size_bytes": len(body),
+            "request_body_prefix": escaped_prefix,
+            "request_body_truncated": len(body) > len(raw_prefix) or len(rendered_prefix) > len(escaped_prefix),
+            "validation_error_count": len(errors),
+            "validation_errors": error_summaries,
+            "validation_errors_truncated": len(errors) > len(error_summaries),
+        },
+    )
+
+
+async def _validation_exception_handler(request: Request, exc: RequestValidationError) -> Response:
+    await _log_validation_exception(request, exc)
+    return await request_validation_exception_handler(request, exc)
 
 
 class _PickleSafeRequestInfo(NamedTuple):
@@ -950,14 +1000,7 @@ def run_webserver(cls) -> Optional[FastAPI]:  # pragma: no cover
         # caller's CLIENT span.
         server.instrument_app_for_telemetry(app)
 
-        @app.exception_handler(RequestValidationError)
-        async def validation_exception_handler(request: Request, exc):
-            print(
-                f"""Hit validation exception! Errors: {json.dumps(exc.errors(), indent=4)}
-Full body: {json.dumps(exc.body, indent=4)}
-"""
-            )
-            return await request_validation_exception_handler(request, exc)
+        app.exception_handler(RequestValidationError)(_validation_exception_handler)
 
         profiling_config = ProfilingMiddlewareConfig.model_validate(global_config_dict)
         if profiling_config.profiling_enabled:

tests/unit_tests/test_server_utils.py

@@ -12,15 +12,19 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+import logging
 import multiprocessing
 import socket
 from concurrent.futures import ProcessPoolExecutor
 from unittest.mock import AsyncMock, MagicMock
 
 from aiohttp import ClientOSError, ClientResponseError, RequestInfo
+from fastapi import Request
+from fastapi.exception_handlers import request_validation_exception_handler
+from fastapi.exceptions import RequestValidationError
 from multidict import CIMultiDict, CIMultiDictProxy
 from omegaconf import OmegaConf
-from pytest import CaptureFixture, MonkeyPatch, raises
+from pytest import CaptureFixture, LogCaptureFixture, MonkeyPatch, mark, raises
 from yarl import URL
 
 import nemo_gym.global_config
@@ -39,7 +43,9 @@
     ServerClient,
     SimpleServer,
     _format_upstream_error_log,
+    _log_validation_exception,
     _make_keepalive_socket_factory,
+    _validation_exception_handler,
     initialize_ray,
     raise_for_status,
 )
@@ -570,6 +576,104 @@ def test_upstream_error_log_has_bounded_body_and_redacted_url(self) -> None:
         assert message.endswith("…")
         assert len(message) < 2200
 
+    @mark.parametrize(
+        "body",
+        [
+            b"",
+            b'{"nested":{"value":"small"}}',
+            b"not-json\nwith-control-\x00",
+            b'{"payload":"' + b"x" * (2 * 1024 * 1024) + b'"}',
+        ],
+        ids=["empty", "small-json", "non-json-control", "multi-megabyte"],
+    )
+    async def test_validation_exception_log_bounds_body_before_rendering(
+        self, body: bytes, caplog: LogCaptureFixture
+    ) -> None:
+        request = MagicMock(spec=Request)
+        request.body = AsyncMock(return_value=body)
+        errors = [
+            {
+                "type": "missing",
+                "loc": ("body", "required_field"),
+                "msg": "Field required",
+                "input": {"large": "value that must not be copied into the log"},
+            }
+        ]
+        exc = RequestValidationError(errors, body={"original": "body"})
+
+        with caplog.at_level(logging.WARNING, logger="nemo_gym.server_utils"):
+            await _log_validation_exception(request, exc)
+
+        record = caplog.records[-1]
+        assert record.request_body_size_bytes == len(body)
+        assert len(record.request_body_prefix) <= nemo_gym.server_utils._VALIDATION_ERROR_LOG_BODY_CHARS
+        assert record.request_body_truncated is (len(body) > nemo_gym.server_utils._VALIDATION_ERROR_LOG_BODY_CHARS)
+        assert record.validation_error_count == 1
+        assert record.validation_errors == [
+            {
+                "type": "missing",
+                "loc": ("body", "required_field"),
+                "msg": "Field required",
+            }
+        ]
+        assert "value that must not be copied into the log" not in str(record.validation_errors)
+        if body == b"not-json\nwith-control-\x00":
+            assert "\n" not in record.request_body_prefix
+            assert "\x00" not in record.request_body_prefix
+            assert "\\n" in record.request_body_prefix
+            assert "\\u0000" in record.request_body_prefix
+
+    async def test_validation_exception_log_bounds_error_count(self, caplog: LogCaptureFixture) -> None:
+        request = MagicMock(spec=Request)
+        request.body = AsyncMock(return_value=b"{}")
+        errors = [
+            {
+                "type": "missing",
+                "loc": ("body", f"field_{index}"),
+                "msg": "Field required",
+                "input": None,
+            }
+            for index in range(nemo_gym.server_utils._VALIDATION_ERROR_LOG_MAX_ERRORS + 5)
+        ]
+
+        with caplog.at_level(logging.WARNING, logger="nemo_gym.server_utils"):
+            await _log_validation_exception(request, RequestValidationError(errors))
+
+        record = caplog.records[-1]
+        assert record.validation_error_count == len(errors)
+        assert len(record.validation_errors) == nemo_gym.server_utils._VALIDATION_ERROR_LOG_MAX_ERRORS
+        assert record.validation_errors_truncated is True
+
+    async def test_validation_exception_logging_failure_does_not_mask_422(self, caplog: LogCaptureFixture) -> None:
+        request = MagicMock(spec=Request)
+        request.body = AsyncMock(side_effect=RuntimeError("body unavailable"))
+        errors = [{"type": "missing", "loc": ("body", "field"), "msg": "Field required", "input": {}}]
+        exc = RequestValidationError(errors, body={"field": None})
+
+        with caplog.at_level(logging.WARNING, logger="nemo_gym.server_utils"):
+            await _log_validation_exception(request, exc)
+
+        record = caplog.records[-1]
+        assert record.getMessage() == "Request validation failed; request body unavailable"
+        assert record.validation_error_count == 1
+        assert exc.errors() == errors
+        assert exc.body == {"field": None}
+
+    async def test_validation_exception_handler_preserves_fastapi_response(self, caplog: LogCaptureFixture) -> None:
+        request = MagicMock(spec=Request)
+        request.body = AsyncMock(return_value=b'{"field":null}')
+        exc = RequestValidationError(
+            [{"type": "missing", "loc": ("body", "required"), "msg": "Field required", "input": None}]
+        )
+        expected = await request_validation_exception_handler(request, exc)
+
+        with caplog.at_level(logging.WARNING, logger="nemo_gym.server_utils"):
+            actual = await _validation_exception_handler(request, exc)
+
+        assert actual.status_code == expected.status_code == 422
+        assert actual.body == expected.body
+        assert actual.headers == expected.headers
+
     async def test_exception_middleware_logs_upstream_error_without_debug(
         self, monkeypatch: MonkeyPatch, capsys: CaptureFixture[str]
     ) -> None: