diff --git a/nemo_gym/server_utils.py b/nemo_gym/server_utils.py index 83f98987ea..d82801eb19 100644 --- a/nemo_gym/server_utils.py +++ b/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: diff --git a/tests/unit_tests/test_server_utils.py b/tests/unit_tests/test_server_utils.py index 92205fc353..9925eb76b6 100644 --- a/tests/unit_tests/test_server_utils.py +++ b/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: