From 17dbb579698ec4a722111c129ec0129bfedf4257 Mon Sep 17 00:00:00 2001 From: oss-agent-shin Date: Thu, 28 May 2026 09:56:54 -0700 Subject: [PATCH 01/11] fix(datadog): chunk on 413 and honor DD_BATCH_SIZE (LIT-3407) [1/3] --- litellm/integrations/datadog/datadog.py | 185 ++++++++++++++++++++---- 1 file changed, 155 insertions(+), 30 deletions(-) diff --git a/litellm/integrations/datadog/datadog.py b/litellm/integrations/datadog/datadog.py index c3e555f6e89b..b961157ad13f 100644 --- a/litellm/integrations/datadog/datadog.py +++ b/litellm/integrations/datadog/datadog.py @@ -41,6 +41,7 @@ ) from litellm.litellm_core_utils.dd_tracing import tracer from litellm.llms.custom_httpx.http_handler import ( + MaskedHTTPStatusError, _get_httpx_client, get_async_httpx_client, httpxSpecialProvider, @@ -68,6 +69,55 @@ ] +def _resolve_dd_batch_size() -> int: + """ + Resolve the per-flush batch size for the Datadog logger. + + Honors the documented ``DD_BATCH_SIZE`` env var (referenced in + ``DD_ERRORS.DATADOG_413_ERROR``) so operators can drop the batch size to + work around oversized payloads. Falls back to ``DD_MAX_BATCH_SIZE`` (1000) + when the env var is unset / invalid / non-positive. Values greater than + ``DD_MAX_BATCH_SIZE`` are clamped down to that ceiling. + """ + raw = os.getenv("DD_BATCH_SIZE") + if not raw: + return DD_MAX_BATCH_SIZE + try: + parsed = int(raw) + except (TypeError, ValueError): + verbose_logger.warning( + "Datadog: ignoring non-integer DD_BATCH_SIZE=%r; falling back to %s", + raw, + DD_MAX_BATCH_SIZE, + ) + return DD_MAX_BATCH_SIZE + if parsed <= 0: + verbose_logger.warning( + "Datadog: ignoring non-positive DD_BATCH_SIZE=%r; falling back to %s", + raw, + DD_MAX_BATCH_SIZE, + ) + return DD_MAX_BATCH_SIZE + return min(parsed, DD_MAX_BATCH_SIZE) + + +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, @@ -128,7 +178,9 @@ def __init__( asyncio.create_task(self.periodic_flush()) self.flush_lock = asyncio.Lock() super().__init__( - **kwargs, flush_lock=self.flush_lock, batch_size=DD_MAX_BATCH_SIZE + **kwargs, + flush_lock=self.flush_lock, + batch_size=_resolve_dd_batch_size(), ) except Exception as e: verbose_logger.exception( @@ -317,56 +369,129 @@ 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. + self.log_queue.extend(batch_to_send) + 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: - self.log_queue = batch_to_send + self.log_queue + self.log_queue.extend(batch_to_send) 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: + verbose_logger.exception( + "%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: From 3923dd11cfd08e3fa0dc767b527f1c9c933256ce Mon Sep 17 00:00:00 2001 From: oss-agent-shin Date: Thu, 28 May 2026 09:56:56 -0700 Subject: [PATCH 02/11] test(datadog): assert new 413 split-and-drop behaviour (LIT-3407) [2/3] --- .../datadog/test_datadog_logger_batching.py | 21 ++++++++++++------- 1 file changed, 14 insertions(+), 7 deletions(-) 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 From 656cde664ff9350ac08c1464930e354a019c216f Mon Sep 17 00:00:00 2001 From: oss-agent-shin Date: Thu, 28 May 2026 09:56:58 -0700 Subject: [PATCH 03/11] test(datadog): regression tests for 413 chunking + DD_BATCH_SIZE (LIT-3407) [3/3] --- .../datadog/test_datadog_413_chunking.py | 272 ++++++++++++++++++ 1 file changed, 272 insertions(+) create mode 100644 tests/test_litellm/integrations/datadog/test_datadog_413_chunking.py 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..c73511288d3a --- /dev/null +++ b/tests/test_litellm/integrations/datadog/test_datadog_413_chunking.py @@ -0,0 +1,272 @@ +""" +LIT-3407 regression tests for DataDog 413 (Payload Too Large) handling and +DD_BATCH_SIZE env-var support. + +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. DD_BATCH_SIZE - referenced in the user-facing error message as the +documented workaround - was never read from the environment. + +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. + * DD_BATCH_SIZE: honored when valid, ignored (with a warning) when invalid. +""" + +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, + _resolve_dd_batch_size, +) +from litellm.llms.custom_httpx.http_handler import MaskedHTTPStatusError +from litellm.types.integrations.datadog import ( + DD_MAX_BATCH_SIZE, + DatadogPayload, +) + + +@pytest.fixture +def datadog_env(monkeypatch): + monkeypatch.setenv("DD_API_KEY", "test_api_key") + monkeypatch.setenv("DD_SITE", "test.datadoghq.com") + monkeypatch.delenv("DD_BATCH_SIZE", raising=False) + + +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() + + # 1 (full batch of 4) + 2 (halves of 2) + 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 == [] + + +# --------------------------------------------------------------------------- # +# DD_BATCH_SIZE env-var (previously documented but never read) # +# --------------------------------------------------------------------------- # +def test_resolve_dd_batch_size_unset_returns_default(): + assert _resolve_dd_batch_size() == DD_MAX_BATCH_SIZE + + +def test_resolve_dd_batch_size_honors_valid_int(monkeypatch): + monkeypatch.setenv("DD_BATCH_SIZE", "50") + assert _resolve_dd_batch_size() == 50 + + +def test_resolve_dd_batch_size_clamps_to_dd_max(monkeypatch): + monkeypatch.setenv("DD_BATCH_SIZE", str(DD_MAX_BATCH_SIZE * 10)) + assert _resolve_dd_batch_size() == DD_MAX_BATCH_SIZE + + +def test_resolve_dd_batch_size_ignores_non_integer(monkeypatch): + monkeypatch.setenv("DD_BATCH_SIZE", "not-a-number") + assert _resolve_dd_batch_size() == DD_MAX_BATCH_SIZE + + +def test_resolve_dd_batch_size_ignores_non_positive(monkeypatch): + monkeypatch.setenv("DD_BATCH_SIZE", "0") + assert _resolve_dd_batch_size() == DD_MAX_BATCH_SIZE + monkeypatch.setenv("DD_BATCH_SIZE", "-5") + assert _resolve_dd_batch_size() == DD_MAX_BATCH_SIZE + + +@pytest.mark.asyncio +async def test_dd_batch_size_env_var_applied_to_logger_init(monkeypatch): + monkeypatch.setenv("DD_API_KEY", "test") + monkeypatch.setenv("DD_SITE", "test.datadoghq.com") + monkeypatch.setenv("DD_BATCH_SIZE", "50") + with patch("asyncio.create_task"): + logger = DataDogLogger() + assert logger.batch_size == 50 + + +@pytest.mark.asyncio +async def test_dd_batch_size_default_when_env_unset(monkeypatch): + monkeypatch.setenv("DD_API_KEY", "test") + monkeypatch.setenv("DD_SITE", "test.datadoghq.com") + monkeypatch.delenv("DD_BATCH_SIZE", raising=False) + with patch("asyncio.create_task"): + logger = DataDogLogger() + assert logger.batch_size == DD_MAX_BATCH_SIZE From 99af899d9bbee9c2e34307a5b3a69aa7bbb149f0 Mon Sep 17 00:00:00 2001 From: oss-agent-shin Date: Thu, 28 May 2026 10:00:10 -0700 Subject: [PATCH 04/11] fix(datadog): chunk on 413 and honor DD_BATCH_SIZE (LIT-3407) [1/3] --- litellm/integrations/datadog/datadog.py | 1 - 1 file changed, 1 deletion(-) diff --git a/litellm/integrations/datadog/datadog.py b/litellm/integrations/datadog/datadog.py index b961157ad13f..e9e99b098195 100644 --- a/litellm/integrations/datadog/datadog.py +++ b/litellm/integrations/datadog/datadog.py @@ -41,7 +41,6 @@ ) from litellm.litellm_core_utils.dd_tracing import tracer from litellm.llms.custom_httpx.http_handler import ( - MaskedHTTPStatusError, _get_httpx_client, get_async_httpx_client, httpxSpecialProvider, From 5964ed02847b5c8d4e79714f276bd06ea2a6587c Mon Sep 17 00:00:00 2001 From: oss-agent-shin Date: Thu, 28 May 2026 10:00:12 -0700 Subject: [PATCH 05/11] test(datadog): assert new 413 split-and-drop behaviour (LIT-3407) [2/3] From 3541ac68ebc17f00bbd2ff69990bffd9c03bcaa4 Mon Sep 17 00:00:00 2001 From: oss-agent-shin Date: Thu, 28 May 2026 10:00:14 -0700 Subject: [PATCH 06/11] test(datadog): regression tests for 413 chunking + DD_BATCH_SIZE (LIT-3407) [3/3] From d4c870eb22955b01189adefc868c79ba5df77b68 Mon Sep 17 00:00:00 2001 From: oss-agent-shin Date: Thu, 28 May 2026 10:06:08 -0700 Subject: [PATCH 07/11] fix(datadog): chunk batch on 413 instead of looping (LIT-3407) [1/4] --- litellm/integrations/datadog/datadog.py | 36 +------------------------ 1 file changed, 1 insertion(+), 35 deletions(-) diff --git a/litellm/integrations/datadog/datadog.py b/litellm/integrations/datadog/datadog.py index e9e99b098195..4e2d09175dad 100644 --- a/litellm/integrations/datadog/datadog.py +++ b/litellm/integrations/datadog/datadog.py @@ -68,38 +68,6 @@ ] -def _resolve_dd_batch_size() -> int: - """ - Resolve the per-flush batch size for the Datadog logger. - - Honors the documented ``DD_BATCH_SIZE`` env var (referenced in - ``DD_ERRORS.DATADOG_413_ERROR``) so operators can drop the batch size to - work around oversized payloads. Falls back to ``DD_MAX_BATCH_SIZE`` (1000) - when the env var is unset / invalid / non-positive. Values greater than - ``DD_MAX_BATCH_SIZE`` are clamped down to that ceiling. - """ - raw = os.getenv("DD_BATCH_SIZE") - if not raw: - return DD_MAX_BATCH_SIZE - try: - parsed = int(raw) - except (TypeError, ValueError): - verbose_logger.warning( - "Datadog: ignoring non-integer DD_BATCH_SIZE=%r; falling back to %s", - raw, - DD_MAX_BATCH_SIZE, - ) - return DD_MAX_BATCH_SIZE - if parsed <= 0: - verbose_logger.warning( - "Datadog: ignoring non-positive DD_BATCH_SIZE=%r; falling back to %s", - raw, - DD_MAX_BATCH_SIZE, - ) - return DD_MAX_BATCH_SIZE - return min(parsed, DD_MAX_BATCH_SIZE) - - def _is_413_error(exc: BaseException) -> bool: """ Detect a Datadog 413 (Payload Too Large) hidden inside an exception. @@ -177,9 +145,7 @@ def __init__( asyncio.create_task(self.periodic_flush()) self.flush_lock = asyncio.Lock() super().__init__( - **kwargs, - flush_lock=self.flush_lock, - batch_size=_resolve_dd_batch_size(), + **kwargs, flush_lock=self.flush_lock, batch_size=DD_MAX_BATCH_SIZE ) except Exception as e: verbose_logger.exception( From 2443b5808d5d71290daea32d35f03d70e1110dd2 Mon Sep 17 00:00:00 2001 From: oss-agent-shin Date: Thu, 28 May 2026 10:06:09 -0700 Subject: [PATCH 08/11] fix(datadog): update DATADOG_413_ERROR to drop phantom DD_BATCH_SIZE hint (LIT-3407) [2/4] --- litellm/types/integrations/datadog.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) 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): From 8a6bce5f464ecd9d2756385a73dd17169af689b7 Mon Sep 17 00:00:00 2001 From: oss-agent-shin Date: Thu, 28 May 2026 10:06:11 -0700 Subject: [PATCH 09/11] test(datadog): assert new 413 split-and-drop behaviour (LIT-3407) [3/4] From 54d96ec47f2ea066dd4485bc2cebbdadf23d2800 Mon Sep 17 00:00:00 2001 From: oss-agent-shin Date: Thu, 28 May 2026 10:06:13 -0700 Subject: [PATCH 10/11] test(datadog): regression tests for 413 chunking (LIT-3407) [4/4] --- .../datadog/test_datadog_413_chunking.py | 70 ++----------------- 1 file changed, 5 insertions(+), 65 deletions(-) diff --git a/tests/test_litellm/integrations/datadog/test_datadog_413_chunking.py b/tests/test_litellm/integrations/datadog/test_datadog_413_chunking.py index c73511288d3a..99a7f99bef3c 100644 --- a/tests/test_litellm/integrations/datadog/test_datadog_413_chunking.py +++ b/tests/test_litellm/integrations/datadog/test_datadog_413_chunking.py @@ -1,6 +1,5 @@ """ -LIT-3407 regression tests for DataDog 413 (Payload Too Large) handling and -DD_BATCH_SIZE env-var support. +LIT-3407 regression tests for DataDog 413 (Payload Too Large) handling. Bug summary ----------- @@ -10,8 +9,7 @@ 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. DD_BATCH_SIZE - referenced in the user-facing error message as the -documented workaround - was never read from the environment. +forever. These tests pin the new behaviour: * Masked 413 exception -> recursive halve-and-retry, single-event drop. @@ -19,7 +17,6 @@ * 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. - * DD_BATCH_SIZE: honored when valid, ignored (with a warning) when invalid. """ from unittest.mock import AsyncMock, patch @@ -28,23 +25,15 @@ import pytest from httpx import Request, Response -from litellm.integrations.datadog.datadog import ( - DataDogLogger, - _is_413_error, - _resolve_dd_batch_size, -) +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 ( - DD_MAX_BATCH_SIZE, - DatadogPayload, -) +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") - monkeypatch.delenv("DD_BATCH_SIZE", raising=False) def _payloads(n): @@ -107,7 +96,7 @@ async def test_masked_413_exception_triggers_chunking_and_drop(datadog_env): await logger.async_send_batch() - # 1 (full batch of 4) + 2 (halves of 2) + 4 (single events) = 7 calls + # 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 == [] @@ -221,52 +210,3 @@ async def test_413_response_without_exception_also_chunks(datadog_env): # 1 full batch + 2 single-event retries = 3 calls assert logger.async_send_compressed_data.await_count == 3 assert logger.log_queue == [] - - -# --------------------------------------------------------------------------- # -# DD_BATCH_SIZE env-var (previously documented but never read) # -# --------------------------------------------------------------------------- # -def test_resolve_dd_batch_size_unset_returns_default(): - assert _resolve_dd_batch_size() == DD_MAX_BATCH_SIZE - - -def test_resolve_dd_batch_size_honors_valid_int(monkeypatch): - monkeypatch.setenv("DD_BATCH_SIZE", "50") - assert _resolve_dd_batch_size() == 50 - - -def test_resolve_dd_batch_size_clamps_to_dd_max(monkeypatch): - monkeypatch.setenv("DD_BATCH_SIZE", str(DD_MAX_BATCH_SIZE * 10)) - assert _resolve_dd_batch_size() == DD_MAX_BATCH_SIZE - - -def test_resolve_dd_batch_size_ignores_non_integer(monkeypatch): - monkeypatch.setenv("DD_BATCH_SIZE", "not-a-number") - assert _resolve_dd_batch_size() == DD_MAX_BATCH_SIZE - - -def test_resolve_dd_batch_size_ignores_non_positive(monkeypatch): - monkeypatch.setenv("DD_BATCH_SIZE", "0") - assert _resolve_dd_batch_size() == DD_MAX_BATCH_SIZE - monkeypatch.setenv("DD_BATCH_SIZE", "-5") - assert _resolve_dd_batch_size() == DD_MAX_BATCH_SIZE - - -@pytest.mark.asyncio -async def test_dd_batch_size_env_var_applied_to_logger_init(monkeypatch): - monkeypatch.setenv("DD_API_KEY", "test") - monkeypatch.setenv("DD_SITE", "test.datadoghq.com") - monkeypatch.setenv("DD_BATCH_SIZE", "50") - with patch("asyncio.create_task"): - logger = DataDogLogger() - assert logger.batch_size == 50 - - -@pytest.mark.asyncio -async def test_dd_batch_size_default_when_env_unset(monkeypatch): - monkeypatch.setenv("DD_API_KEY", "test") - monkeypatch.setenv("DD_SITE", "test.datadoghq.com") - monkeypatch.delenv("DD_BATCH_SIZE", raising=False) - with patch("asyncio.create_task"): - logger = DataDogLogger() - assert logger.batch_size == DD_MAX_BATCH_SIZE From 11a5e57b2923bbc2a143e00d0cd7edd556f9b8ef Mon Sep 17 00:00:00 2001 From: oss-agent-shin Date: Thu, 28 May 2026 10:09:29 -0700 Subject: [PATCH 11/11] fix(datadog): Greptile feedback - .error not .exception outside try, prepend on requeue (LIT-3407) --- litellm/integrations/datadog/datadog.py | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/litellm/integrations/datadog/datadog.py b/litellm/integrations/datadog/datadog.py index 4e2d09175dad..9ebdfeb9a3d1 100644 --- a/litellm/integrations/datadog/datadog.py +++ b/litellm/integrations/datadog/datadog.py @@ -398,7 +398,9 @@ async def _send_batch_with_413_split(self, batch_to_send: List) -> None: await self._handle_413_split(batch_to_send) return # Non-413 transport error - preserve events for the next flush. - self.log_queue.extend(batch_to_send) + # 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()}" ) @@ -419,7 +421,8 @@ async def _send_batch_with_413_split(self, batch_to_send: List) -> None: f"Response from datadog API status_code: {response.status_code}, text: {response.text}" ) except Exception as e: - self.log_queue.extend(batch_to_send) + # 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()}" ) @@ -439,7 +442,10 @@ async def _send_batch_with_413_split(self, batch_to_send: List) -> None: 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: - verbose_logger.exception( + # 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 "