fix: drain datadog batches safely

[emerzon]

Summary

• detach and drain the Datadog batch queue before sending so threshold flushes do not keep growing memory or resend old events
• route threshold-based flushes through the logger flush lock and requeue unsent events on send failures
• add focused regression tests for concurrent appends during flush and failure-hook threshold flushing

Fixes #25660

[vercel]

The latest updates on your projects. Learn more about Vercel for GitHub.

Project	Deployment	Actions	Updated (UTC)
litellm	Ready	Preview, Comment	Apr 14, 2026 1:59am

[codspeed-hq]

Merging this PR will not alter performance

✅ 16 untouched benchmarks

---

_{Comparing emerzon:fix/datadog-log-queue-leak (c457a68) with main (8427534)}

[greptile-apps]

Greptile Summary

This PR fixes memory growth and event loss in the Datadog logger's batching path by detaching the queue before sending (batch_to_send = self.log_queue[:] + self.log_queue = []), requeuing events on 413 and exceptions, and adding a lock-protected flush_queue override so threshold flushes are serialized with the periodic flush. Both issues highlighted in prior review threads — missing import time and silent event loss on 413 — are resolved here.

Confidence Score: 5/5

Safe to merge — both prior P0/P1 issues are resolved and no new defects introduced.

All previously flagged blocking issues (missing import time, silent event loss on 413) are fixed. The batch-detach pattern, requeue logic, and lock-protected flush_queue override are logically correct. The only remaining finding is a theoretical P2 edge case (batch_to_send unbound before assignment in the exception handler) that requires an already-broken logger state to trigger.

No files require special attention.

Important Files Changed

Filename	Overview
litellm/integrations/datadog/datadog.py	Adds safe batch detach (copy+clear before send), 413/exception requeue, lock-protected flush_queue override, and missing import time; threshold flushes now correctly serialized via the lock.
tests/test_litellm/integrations/datadog/test_datadog_logger_batching.py	New mock-only test file covering concurrent appends during flush, 413/exception requeue, flush_queue lock guard, and last_flush_time update logic — all tests are well-isolated with no real network calls.

Sequence Diagram

sequenceDiagram
    participant H as EventHook / _log_async_event
    participant FQ as flush_queue()
    participant L as flush_lock
    participant ASB as async_send_batch()
    participant DD as Datadog API

    H->>H: log_queue.append(event)
    H->>H: len(queue) >= batch_size?
    H->>FQ: await flush_queue()
    FQ->>L: async with flush_lock
    L-->>FQ: acquired
    FQ->>FQ: if log_queue not empty
    FQ->>ASB: await async_send_batch()
    ASB->>ASB: batch_to_send = log_queue[:]
    ASB->>ASB: log_queue = []
    ASB->>DD: POST /api/v2/logs (batch_to_send)
    alt 202 OK
        DD-->>ASB: 202 Accepted
        ASB-->>FQ: returns (log_queue empty)
        FQ->>FQ: last_flush_time = time.time()
    else 413 Too Large
        DD-->>ASB: 413
        ASB->>ASB: log_queue = batch_to_send + log_queue
        ASB-->>FQ: returns (log_queue non-empty)
        FQ->>FQ: last_flush_time NOT updated
    else Exception
        ASB->>ASB: log_queue = batch_to_send + log_queue
        ASB-->>FQ: returns (log_queue non-empty)
        FQ->>FQ: last_flush_time NOT updated
    end
    FQ->>L: release flush_lock

Loading

_{Reviews (6): Last reviewed commit: "test: use sync mock for datadog payload ..." | Re-trigger Greptile}

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.

📢 Thoughts on this report? Let us know!

[Copilot]

Pull request overview

This PR improves the safety and correctness of the Datadog logger’s batching/flush behavior under concurrency by draining the queue before sending and routing threshold flushes through the batch flush lock, with regression tests to cover concurrent appends and threshold flushing from the failure hook.

Changes:

• Route threshold-triggered flushes (async_post_call_failure_hook, _log_async_event) through flush_queue() to ensure flush-lock serialization.
• Detach (copy + clear) the current Datadog batch before sending, so events appended during the send are not sent twice or lost from the active queue.
• Add focused async tests for “append during send” and “failure-hook triggers flush_queue”.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.

File	Description
litellm/integrations/datadog/datadog.py	Updates batching/flush flow to drain queue before send, requeue on exceptions, and route threshold flushes through flush_queue().
tests/test_litellm/integrations/datadog/test_datadog_logger_batching.py	Adds regression tests validating concurrent appends during send and ensuring failure-hook threshold flush uses flush_queue().

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[emerzon]

Addressed the review feedback in the latest signed commit.

Changes made:

• re-queued batch_to_send on the 413 path so drained Datadog events are not lost
• aligned flush_queue() with the existing batch logger conventions by using a Datadog log prefix and time.time() for last_flush_time
• added a focused regression test covering the 413 requeue path

Verification:

• poetry run pytest tests/test_litellm/integrations/datadog/test_datadog_logger_batching.py -q
• 3 passed

Latest commit: af3eb66f8b