mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-07 08:26:10 +00:00
fix(newrelic): drop only records queued when the drain began, not mid-drain arrivals
This commit is contained in:
parent
5e0d19b4a8
commit
800365cbef
2 changed files with 35 additions and 8 deletions
|
|
@ -207,17 +207,16 @@ class NewRelicMetricsLogger(CustomBatchLogger):
|
|||
of racing it. Each pass attempts the whole current queue in
|
||||
``batch_size`` chunks, unlike the periodic path it does not stop at the
|
||||
first failing chunk, so a persistently failing head never starves the
|
||||
tail, and a record appended mid-drain is snapshotted and delivered by a
|
||||
later pass. Only after ``_MAX_DRAIN_PASSES`` against a permanently
|
||||
failing destination is the remainder dropped, and then only the records
|
||||
this drain actually tried: a record a callback appended after our last
|
||||
snapshot is left for its own serialized drain, so it is never wiped
|
||||
un-tried and never stranded.
|
||||
tail. Only after ``_MAX_DRAIN_PASSES`` against a permanently failing
|
||||
destination is the remainder dropped, and then only the records that were
|
||||
queued when this drain began, so every dropped record got the full retry
|
||||
budget: a record a callback appended mid-drain is not in that snapshot,
|
||||
so it is left for its own serialized drain rather than dropped after
|
||||
fewer attempts, and is never stranded.
|
||||
"""
|
||||
async with self._drain_lock:
|
||||
attempted: tuple[NewRelicMetricRecord, ...] = () # rebind-ok: each pass's pre-attempt snapshot
|
||||
attempted: Final = tuple(self.log_queue)
|
||||
for _pass in range(NEWRELIC_METRICS_MAX_DRAIN_PASSES):
|
||||
attempted = tuple(self.log_queue)
|
||||
await self._drain_flush_once()
|
||||
if not self.log_queue:
|
||||
return
|
||||
|
|
|
|||
|
|
@ -716,6 +716,34 @@ async def test_terminal_drop_leaves_untried_late_arrival_for_next_drain():
|
|||
assert logger.log_queue == [late_record], "the un-tried late arrival is left for its own drain, not dropped"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_record_appended_on_an_early_pass_is_not_dropped_short_of_the_retry_budget():
|
||||
"""A record a callback appends during an early drain pass entered the queue
|
||||
after this drain's snapshot, so it has not seen the full retry budget. The
|
||||
terminal drop must clear only records queued when the drain began, leaving
|
||||
the early-pass arrival for its own serialized drain instead of dropping it
|
||||
after fewer than the configured attempts."""
|
||||
logger = _make_logger()
|
||||
logger.stop()
|
||||
early_record = _record(model="early-pass-arrival")
|
||||
posts = {"n": 0}
|
||||
|
||||
async def _fail_and_append_on_first_pass(url, data=None, headers=None, **kw):
|
||||
posts["n"] += 1
|
||||
# One record queued at start means the first pass's post is the 1st;
|
||||
# append during it, before this drain's later passes.
|
||||
if posts["n"] == 1:
|
||||
logger.log_queue.append(early_record)
|
||||
resp = _response(503)
|
||||
raise HTTPStatusError("err", request=resp.request, response=resp)
|
||||
|
||||
with patch("asyncio.sleep", new=AsyncMock()):
|
||||
logger.async_client.post = _fail_and_append_on_first_pass
|
||||
logger.log_queue.append(_record(model="doomed"))
|
||||
await logger._drain_with_retry()
|
||||
assert logger.log_queue == [early_record], "the early-pass arrival is left for its own drain, not dropped short"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_stop_drains_are_serialized():
|
||||
"""A callback that appends to a stopped logger and starts its own drain must
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue