diff --git a/litellm/integrations/datadog/datadog.py b/litellm/integrations/datadog/datadog.py index c3e555f6e89b..9ebdfeb9a3d1 100644 --- a/litellm/integrations/datadog/datadog.py +++ b/litellm/integrations/datadog/datadog.py @@ -68,6 +68,23 @@ ] +def _is_413_error(exc: BaseException) -> bool: + """ + Detect a Datadog 413 (Payload Too Large) hidden inside an exception. + + The async httpx handler used by this logger converts a 413 Response into + a ``MaskedHTTPStatusError`` (subclass of ``httpx.HTTPStatusError``) via + ``_raise_masked_async_error``, which means the response never reaches + ``async_send_batch`` as a return value. We pull the status code off the + exception's ``.response`` attribute so the caller can still recognise + and chunk the offending batch. + """ + response = getattr(exc, "response", None) + if response is None: + return False + return getattr(response, "status_code", None) == 413 + + class DataDogLogger( CustomBatchLogger, AdditionalLoggingUtils, @@ -317,56 +334,135 @@ async def async_send_batch(self): DD Ref: https://docs.datadoghq.com/api/latest/logs/ + Behaviour on a 413 (Payload Too Large) response: + Datadog enforces a 5 MB uncompressed payload limit per call. Prior + to LIT-3407 a 413 was handled by re-queueing the entire oversized + batch unchanged, which caused the same batch to be retried every + flush interval (and never delivered). We now split the offending + batch in half and retry each half independently; if a single + event still cannot fit we drop it with a verbose error log to + avoid an infinite retry loop. + Raises: Raises a NON Blocking verbose_logger.exception if an error occurs """ - try: - if not self.log_queue: - verbose_logger.exception("Datadog: log_queue does not exist") - return + if not self.log_queue: + verbose_logger.exception("Datadog: log_queue does not exist") + return + + batch_to_send = self.log_queue[:] + self.log_queue = [] - batch_to_send = self.log_queue[:] - self.log_queue = [] + verbose_logger.debug( + "Datadog - about to flush %s events on %s", + len(batch_to_send), + self.intake_url, + ) + if self.is_mock_mode: verbose_logger.debug( - "Datadog - about to flush %s events on %s", - len(batch_to_send), - self.intake_url, + "[DATADOG MOCK] Mock mode enabled - API calls will be intercepted" ) - if self.is_mock_mode: - verbose_logger.debug( - "[DATADOG MOCK] Mock mode enabled - API calls will be intercepted" - ) + try: + await self._send_batch_with_413_split(batch_to_send) + except Exception as e: + # Defence in depth: _send_batch_with_413_split already handles + # send failures inline and re-queues. We only reach here on + # programmer error inside the helper; preserve the original batch + # so events are not lost. + self.log_queue = batch_to_send + self.log_queue + verbose_logger.exception( + f"Datadog Error sending batch API - {str(e)}\n{traceback.format_exc()}" + ) + async def _send_batch_with_413_split(self, batch_to_send: List) -> None: + """ + Send a batch, recursively splitting on 413 (Payload Too Large). + + - On 413 with len(batch) > 1: split in half, retry each half. + - On 413 with len(batch) == 1: drop the single event with a verbose + error log (it cannot fit under Datadog's 5 MB limit on its own; + re-queueing would loop forever). + - On any other send failure: re-queue the slice for the next flush + attempt (preserves prior behaviour). + - On 2xx: log debug success. + """ + if not batch_to_send: + return + + try: response = await self.async_send_compressed_data(batch_to_send) - if response.status_code == 413: - verbose_logger.exception(DD_ERRORS.DATADOG_413_ERROR.value) - self.log_queue = batch_to_send + self.log_queue + except Exception as e: + if _is_413_error(e): + await self._handle_413_split(batch_to_send) return + # Non-413 transport error - preserve events for the next flush. + # Prepend so retried events stay ahead of newly enqueued ones + # (matches the legacy ordering Greptile flagged on PR #29183). + self.log_queue = batch_to_send + self.log_queue + verbose_logger.exception( + f"Datadog Error sending batch API - {str(e)}\n{traceback.format_exc()}" + ) + return + + # Defensive: handle the case where the underlying httpx client returns + # a 413 Response instead of raising. The masked httpx handler currently + # raises, but this guards against future refactors / sync clients / + # mock clients. + if getattr(response, "status_code", None) == 413: + await self._handle_413_split(batch_to_send) + return + try: response.raise_for_status() if response.status_code != 202: raise Exception( f"Response from datadog API status_code: {response.status_code}, text: {response.text}" ) - - if self.is_mock_mode: - verbose_logger.debug( - f"[DATADOG MOCK] Batch of {len(batch_to_send)} events successfully mocked" - ) - else: - verbose_logger.debug( - "Datadog: Response from datadog API status_code: %s, text: %s", - response.status_code, - response.text, - ) - except Exception as e: + # Prepend to keep retried events ahead of newly enqueued ones. self.log_queue = batch_to_send + self.log_queue verbose_logger.exception( f"Datadog Error sending batch API - {str(e)}\n{traceback.format_exc()}" ) + return + + if self.is_mock_mode: + verbose_logger.debug( + f"[DATADOG MOCK] Batch of {len(batch_to_send)} events successfully mocked" + ) + else: + verbose_logger.debug( + "Datadog: Response from datadog API status_code: %s, text: %s", + response.status_code, + response.text, + ) + + async def _handle_413_split(self, batch_to_send: List) -> None: + """Split a 413-rejected batch in half and retry. Drop single-event 413s.""" + if len(batch_to_send) <= 1: + # Not inside an `except` block - use .error so verbose_logger + # doesn't print "NoneType: None" for a non-existent traceback + # (Greptile review feedback on PR #29183). + verbose_logger.error( + "%s Single event still exceeds Datadog's 5MB limit; dropping it " + "to avoid an infinite retry loop. Disable request/response body " + "logging via `litellm.turn_off_message_logging = True` or lower " + "the event size to recover.", + DD_ERRORS.DATADOG_413_ERROR.value, + ) + return + mid = len(batch_to_send) // 2 + verbose_logger.warning( + "Datadog: 413 Payload Too Large on batch of %s events; " + "splitting into halves of %s + %s and retrying.", + len(batch_to_send), + mid, + len(batch_to_send) - mid, + ) + await self._send_batch_with_413_split(batch_to_send[:mid]) + await self._send_batch_with_413_split(batch_to_send[mid:]) async def flush_queue(self): if self.flush_lock is None: diff --git a/litellm/types/integrations/datadog.py b/litellm/types/integrations/datadog.py index 89faac278301..f707f411f548 100644 --- a/litellm/types/integrations/datadog.py +++ b/litellm/types/integrations/datadog.py @@ -31,7 +31,14 @@ class DataDogStatus(str, Enum): class DD_ERRORS(Enum): - DATADOG_413_ERROR = "Datadog API Error - Payload too large (batch is above 5MB uncompressed). If you want this logged either disable request/response logging or set `DD_BATCH_SIZE=50`" + DATADOG_413_ERROR = ( + "Datadog API Error - Payload too large (batch is above 5MB " + "uncompressed). The batch is split in half and retried; single " + "events that still exceed the 5MB limit are dropped. To recover, " + "disable request/response body logging via " + "`litellm.turn_off_message_logging = True` or " + "`DatadogInitParams(turn_off_message_logging=True)`." + ) class DatadogInitParams(StandardCustomLoggerInitParams): diff --git a/tests/test_litellm/integrations/datadog/test_datadog_413_chunking.py b/tests/test_litellm/integrations/datadog/test_datadog_413_chunking.py new file mode 100644 index 000000000000..99a7f99bef3c --- /dev/null +++ b/tests/test_litellm/integrations/datadog/test_datadog_413_chunking.py @@ -0,0 +1,212 @@ +""" +LIT-3407 regression tests for DataDog 413 (Payload Too Large) handling. + +Bug summary +----------- +async_send_compressed_data sits on top of the LiteLLM async httpx handler, +which converts any 4xx into a MaskedHTTPStatusError before returning. So the +production codepath for a 413 from Datadog is an *exception*, not a Response +with status_code == 413 - the old `if response.status_code == 413` branch +inside async_send_batch was unreachable, the bare `except` re-queued the +whole oversized batch, and every flush interval replayed the same 413 +forever. + +These tests pin the new behaviour: + * Masked 413 exception -> recursive halve-and-retry, single-event drop. + * Direct 413 Response -> same recursive halve-and-retry, single-event drop. + * Partial split: first half 413, second half 202 -> first half is dropped + after a 1-event recurse, second half is delivered in one call. + * Non-413 transport errors -> events are preserved for retry. +""" + +from unittest.mock import AsyncMock, patch + +import httpx +import pytest +from httpx import Request, Response + +from litellm.integrations.datadog.datadog import DataDogLogger, _is_413_error +from litellm.llms.custom_httpx.http_handler import MaskedHTTPStatusError +from litellm.types.integrations.datadog import DatadogPayload + + +@pytest.fixture +def datadog_env(monkeypatch): + monkeypatch.setenv("DD_API_KEY", "test_api_key") + monkeypatch.setenv("DD_SITE", "test.datadoghq.com") + + +def _payloads(n): + return [ + DatadogPayload( + ddsource="litellm", + ddtags="env:test", + hostname="host", + message=f'{{"event": {i}}}', + service="svc", + status="info", + ) + for i in range(n) + ] + + +def _masked_413(): + """Construct the MaskedHTTPStatusError(413) the production httpx path raises.""" + req = Request("POST", "https://intake-test.datadoghq.com/api/v2/logs") + resp = Response(413, request=req, text="Payload Too Large") + orig = httpx.HTTPStatusError("413 Payload Too Large", request=req, response=resp) + return MaskedHTTPStatusError(orig, message="Payload Too Large", text="Payload Too Large") + + +# --------------------------------------------------------------------------- # +# _is_413_error # +# --------------------------------------------------------------------------- # +def test_is_413_error_detects_masked_httpx_exception(): + assert _is_413_error(_masked_413()) is True + + +def test_is_413_error_returns_false_for_non_413_status(): + req = Request("POST", "https://example.com") + resp = Response(500, request=req, text="boom") + exc = httpx.HTTPStatusError("500", request=req, response=resp) + assert _is_413_error(exc) is False + + +def test_is_413_error_returns_false_for_unrelated_exception(): + assert _is_413_error(RuntimeError("not http")) is False + + +# --------------------------------------------------------------------------- # +# Masked-exception 413 path (the actual production path) # +# --------------------------------------------------------------------------- # +@pytest.mark.asyncio +async def test_masked_413_exception_triggers_chunking_and_drop(datadog_env): + """ + Mirror production: async_send_compressed_data raises MaskedHTTPStatusError(413) + instead of returning a Response. Before LIT-3407 this raise fell through to + the bare `except` and the whole batch was re-queued -> infinite retry loop. + The fix splits the batch recursively until single events are reached, then + drops those single events with a verbose error log. + """ + with patch("asyncio.create_task"): + logger = DataDogLogger() + + logger.log_queue = _payloads(4) + logger.async_send_compressed_data = AsyncMock(side_effect=_masked_413()) + + await logger.async_send_batch() + + # Recursive halving of 4 events: 1 full + 2 halves + 4 single events = 7 calls. + assert logger.async_send_compressed_data.await_count == 7 + # Queue is fully drained - no infinite retry. + assert logger.log_queue == [] + + +@pytest.mark.asyncio +async def test_masked_413_does_not_loop_on_subsequent_flushes(datadog_env): + """ + Sanity: after a 413 flush drains the queue, subsequent flushes do not + re-flush the same oversize batch (which was the LIT-3407 symptom). + """ + with patch("asyncio.create_task"): + logger = DataDogLogger() + + logger.log_queue = _payloads(2) + logger.async_send_compressed_data = AsyncMock(side_effect=_masked_413()) + + await logger.async_send_batch() + first_calls = logger.async_send_compressed_data.await_count + assert logger.log_queue == [] + + # Nothing new in queue -> next flush should be a no-op send. + await logger.async_send_batch() + assert logger.async_send_compressed_data.await_count == first_calls + + +@pytest.mark.asyncio +async def test_413_split_partial_success_only_failing_slice_chunks(datadog_env): + """ + First-half 413, second-half 202. + First half (events 0, 1) is split into [0] and [1] and dropped. + Second half (events 2, 3) is delivered in a single call. + """ + with patch("asyncio.create_task"): + logger = DataDogLogger() + + logger.log_queue = _payloads(4) + + delivered_batches = [] + call_count = {"n": 0} + + async def fake_send(batch): + call_count["n"] += 1 + n = call_count["n"] + if n == 1: + # Full batch of 4 -> 413 (overall too big). + raise _masked_413() + if n == 2: + # First half [event0, event1] -> 413. + raise _masked_413() + if n == 3: + # event 0 alone -> 413 (dropped). + raise _masked_413() + if n == 4: + # event 1 alone -> 413 (dropped). + raise _masked_413() + if n == 5: + # Second half [event2, event3] -> 202 (delivered). + delivered_batches.append([p["message"] for p in batch]) + return Response( + 202, request=Request("POST", "https://example.com"), text="Accepted" + ) + raise AssertionError(f"unexpected call #{n}") + + logger.async_send_compressed_data = AsyncMock(side_effect=fake_send) + + await logger.async_send_batch() + + assert logger.async_send_compressed_data.await_count == 5 + assert logger.log_queue == [] + assert delivered_batches == [['{"event": 2}', '{"event": 3}']] + + +@pytest.mark.asyncio +async def test_non_413_transport_error_preserves_full_batch(datadog_env): + """Non-413 errors must still re-queue the entire slice (current contract).""" + with patch("asyncio.create_task"): + logger = DataDogLogger() + + logger.log_queue = _payloads(3) + logger.async_send_compressed_data = AsyncMock(side_effect=RuntimeError("boom")) + + await logger.async_send_batch() + + assert logger.async_send_compressed_data.await_count == 1 + assert [p["message"] for p in logger.log_queue] == [ + '{"event": 0}', + '{"event": 1}', + '{"event": 2}', + ] + + +@pytest.mark.asyncio +async def test_413_response_without_exception_also_chunks(datadog_env): + """ + Defensive coverage for clients that return a 413 Response instead of + raising (sync clients, mocks, future httpx refactors). + """ + with patch("asyncio.create_task"): + logger = DataDogLogger() + + logger.log_queue = _payloads(2) + logger.async_send_compressed_data = AsyncMock( + return_value=Response( + 413, request=Request("POST", "https://example.com"), text="too big" + ) + ) + + await logger.async_send_batch() + + # 1 full batch + 2 single-event retries = 3 calls + assert logger.async_send_compressed_data.await_count == 3 + assert logger.log_queue == [] diff --git a/tests/test_litellm/integrations/datadog/test_datadog_logger_batching.py b/tests/test_litellm/integrations/datadog/test_datadog_logger_batching.py index e4d7227cc88d..7cd7ebbe5c28 100644 --- a/tests/test_litellm/integrations/datadog/test_datadog_logger_batching.py +++ b/tests/test_litellm/integrations/datadog/test_datadog_logger_batching.py @@ -75,7 +75,16 @@ async def test_failure_hook_threshold_flush_uses_flush_queue(datadog_env): @pytest.mark.asyncio -async def test_async_send_batch_requeues_events_on_413(datadog_env): +async def test_async_send_batch_splits_and_drops_on_413_response(datadog_env): + """ + LIT-3407 - When Datadog returns 413 (Payload Too Large), the batch must be + split in half and retried. If every event still 413s individually, drop + them with a verbose error log instead of re-queueing forever. + + Pre-LIT-3407 behaviour (re-queue whole batch unchanged) caused an + infinite retry loop because the same oversize payload would fire 413 on + every subsequent flush, and no events were ever delivered. + """ with patch("asyncio.create_task"): logger = DataDogLogger() @@ -101,12 +110,10 @@ async def test_async_send_batch_requeues_events_on_413(datadog_env): await logger.async_send_batch() - assert logger.async_send_compressed_data.await_count == 1 - assert len(logger.log_queue) == 2 - assert [event["message"] for event in logger.log_queue] == [ - '{"event": 0}', - '{"event": 1}', - ] + # 1 attempt for the full batch + 2 attempts for the two single-event halves + assert logger.async_send_compressed_data.await_count == 3 + # Single-event 413s are dropped (would loop forever otherwise). + assert logger.log_queue == [] @pytest.mark.asyncio