Check for existing issues

• I have searched the existing issues and checked that my issue is not a duplicate.

What happened?

DataDogLogger accumulates an unbounded in-memory queue between periodic flushes, and re-sends every previously-seen event on every request once the batch threshold is crossed. This causes both a memory growth proportional to throughput × flush interval and quadratic duplicate API calls to Datadog.

Severity

Critical — on the hot path for every LLM response when Datadog logging is enabled. At 1,000 req/s this consumes ~18 MB per 5-second flush cycle; at 10,000 req/s ~150 MB; at 100,000 req/s ~1.5 GB.

Affected files

• litellm/integrations/datadog/datadog.py — lines 293–310, 311–365, 418–432
• litellm/integrations/custom_batch_logger.py — lines 44–55 (has the only correct log_queue.clear())

Root cause

_log_async_event() appends each event to self.log_queue and then, when len(log_queue) >= batch_size (default DD_MAX_BATCH_SIZE = 1000), calls async_send_batch() directly:

# datadog.py:426-432
self.log_queue.append(dd_payload)

if len(self.log_queue) >= self.batch_size:
    await self.async_send_batch()   # ← sends queue but NEVER clears it

async_send_batch() transmits the queue to Datadog but contains no self.log_queue.clear() call:

# datadog.py:311-362
async def async_send_batch(self):
    try:
        if not self.log_queue:
            ...
            return

        response = await self.async_send_compressed_data(self.log_queue)  # sends all
        response.raise_for_status()
        ...
        # ← log_queue.clear() is MISSING here
    except Exception as e:
        verbose_logger.exception(...)

The queue is only cleared inside the base class flush_queue(), called by the background periodic_flush() task every flush_interval seconds (default 5 s):

# custom_batch_logger.py:44-55
async def flush_queue(self):
    async with self.flush_lock:
        if self.log_queue:
            await self.async_send_batch()
            self.log_queue.clear()    # ← the ONLY place that clears

The same defect is present in async_post_call_failure_hook() (lines 301–304), which also calls async_send_batch() directly after appending to the queue.

How to reproduce

1. Configure a LiteLLM proxy with Datadog logging enabled.
2. Send ≥ 1,001 requests within a single flush_interval window (default 5 s).
3. Observe that:
   • Memory: the proxy's RSS grows proportionally to throughput × flush_interval.
   • Datadog: events with index 1,000 onward appear multiple times in Datadog (event at index 1,000 appears (T × F − batch_size) times).

Impact

Call chain (static trace)

litellm proxy HTTP request
  → FastAPI route handler (proxy_server.py)
    → litellm.acompletion()
      → Logging.success_handler()
        → DataDogLogger.async_log_success_event()    [datadog.py:184]
          → DataDogLogger._log_async_event()         [datadog.py:418]
            → self.log_queue.append(dd_payload)      [datadog.py:426]  ← GROWS
            → self.async_send_batch()                [datadog.py:432]  ← NO CLEAR

Parallel failure path:
  → DataDogLogger.async_post_call_failure_hook()     [datadog.py:221]
    → self.log_queue.append(dd_payload)              [datadog.py:301]  ← GROWS
    → self.async_send_batch()                        [datadog.py:304]  ← NO CLEAR

Proposed fix

Option A (recommended) — Add self.log_queue.clear() inside async_send_batch() in datadog.py, after a successful send. This mirrors the correct pattern already used in SlackAlerting:

# datadog.py — DataDogLogger.async_send_batch()
async def async_send_batch(self):
    try:
        if not self.log_queue:
            verbose_logger.exception("Datadog: log_queue does not exist")
            return

        response = await self.async_send_compressed_data(self.log_queue)
        if response.status_code == 413:
            verbose_logger.exception(DD_ERRORS.DATADOG_413_ERROR.value)
            return

        response.raise_for_status()
        if response.status_code != 202:
            raise Exception(...)

        self.log_queue.clear()   # ← ADD THIS after successful send
        verbose_logger.debug(...)
    except Exception as e:
        verbose_logger.exception(...)

Option B — Replace direct async_send_batch() calls in _log_async_event() and async_post_call_failure_hook() with flush_queue(), which already handles both sending and clearing under the flush_lock:

# datadog.py:431-432 (and line 303-304)
if len(self.log_queue) >= self.batch_size:
    await self.flush_queue()   # ← uses the lock AND clears

Option A is preferred because it also closes a race between _log_async_event() (which bypasses flush_lock) and the concurrent periodic_flush() task — both can hold a reference to the same queue contents simultaneously.

Reference

• SlackAlerting.async_send_batch() (correct pattern): clears the queue inside async_send_batch().
• CustomBatchLogger.flush_queue() (custom_batch_logger.py:44–55): the base-class flush that currently does the only safe clear.

Environment

• Component: litellm/integrations/datadog/
• Trigger condition: Datadog logging enabled + throughput ≥ DD_MAX_BATCH_SIZE (1000) events within a single flush_interval (default 5 s)
• Python: 3.10+
• LiteLLM versions affected: all versions with DataDogLogger

What part of LiteLLM is this about?

Proxy

What LiteLLM version are you on ?

v1.83.7

Twitter / LinkedIn details

No response