scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-13 23:11:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 101

Merge pull request #40912 from BerriAI/litellm_logging_worker_timeout_summary

Browse source 
fix(logging): log one bounded summary for a burst of timed-out LoggingWorker callbacks
This commit is contained in:
Mateo Wang 2026-09-12 19:12:22 -07:00 • committed by GitHub
commit e240997529
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 262 additions and 40 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
litellm/constants.py
View file

@ -571,6 +571,7 @@ ANTHROPIC_MESSAGES_MAX_DETACHED_STREAM_DRAINS: Final = int(
LOGGING_WORKER_CONCURRENCY: Final = int(os.getenv("LOGGING_WORKER_CONCURRENCY", 100)) # Must be above 0
LOGGING_WORKER_MAX_QUEUE_SIZE: Final = int(os.getenv("LOGGING_WORKER_MAX_QUEUE_SIZE", 50_000))
LOGGING_WORKER_MAX_TIME_PER_COROUTINE: Final = float(os.getenv("LOGGING_WORKER_MAX_TIME_PER_COROUTINE", 20.0))
LOGGING_WORKER_TIMEOUT_SUMMARY_WINDOW_SECONDS: Final = 5.0
LOGGING_WORKER_CLEAR_PERCENTAGE: Final = int(
os.getenv("LOGGING_WORKER_CLEAR_PERCENTAGE", 50)
) # Percentage of queue to clear (default: 50%)

60
litellm/litellm_core_utils/logging_worker.py
View file

@ -18,11 +18,16 @@ from litellm.constants import (
LOGGING_WORKER_CONCURRENCY,
LOGGING_WORKER_MAX_QUEUE_SIZE,
LOGGING_WORKER_MAX_TIME_PER_COROUTINE,
LOGGING_WORKER_TIMEOUT_SUMMARY_WINDOW_SECONDS,
MAX_ITERATIONS_TO_CLEAR_QUEUE,
MAX_TIME_TO_CLEAR_QUEUE,
)
def _coroutine_name(coroutine: Coroutine) -> str:
return getattr(coroutine, "__qualname__", None) or getattr(coroutine, "__name__", None) or type(coroutine).__name__
class LoggingTask(TypedDict):
"""
A logging task with its associated context to ensure logging is executed in
@ -47,10 +52,12 @@ class LoggingWorker:
timeout: float = LOGGING_WORKER_MAX_TIME_PER_COROUTINE,
max_queue_size: int = LOGGING_WORKER_MAX_QUEUE_SIZE,
concurrency: int = LOGGING_WORKER_CONCURRENCY,
timeout_summary_window: float = LOGGING_WORKER_TIMEOUT_SUMMARY_WINDOW_SECONDS,
):
self.timeout = timeout
self.max_queue_size = max_queue_size
self.concurrency = concurrency
self.timeout_summary_window = timeout_summary_window
self._queue: asyncio.Queue[LoggingTask] | None = None
self._worker_task: asyncio.Task | None = None
self._running_tasks: set[asyncio.Task] = set()
@ -59,6 +66,10 @@ class LoggingWorker:
self._bound_loop: asyncio.AbstractEventLoop | None = None
self._last_aggressive_clear_time: float = 0.0
self._aggressive_clear_in_progress: bool = False
self._timeout_total: int = 0
self._timeout_burst_count: int = 0
self._timeout_last_callback: str | None = None
self._timeout_summary_task: asyncio.Task | None = None
# Register cleanup handler to flush remaining events on exit
atexit.register(self._flush_on_exit)
@ -136,6 +147,8 @@ class LoggingWorker:
self._sem = None
self._worker_task = None
self._running_tasks.clear()
self._timeout_summary_task = None
self._timeout_burst_count = 0
self._queue = new_queue
self._bound_loop = current_loop
return
@ -156,12 +169,15 @@ class LoggingWorker:
"""Runs the logging task and handles cleanup. Releases semaphore when done."""
try:
if self._queue is not None:
# Run the coroutine in its original context
callback_task: Final = task["context"].run(asyncio.create_task, task["coroutine"])
try:
# Run the coroutine in its original context
await asyncio.wait_for(
task["context"].run(asyncio.create_task, task["coroutine"]),
timeout=self.timeout,
)
await asyncio.wait_for(callback_task, timeout=self.timeout)
except asyncio.TimeoutError as e:
if callback_task.cancelled():
self._record_callback_timeout(task["coroutine"])
else:
verbose_logger.exception("LoggingWorker error: %s", e)
except Exception as e:
verbose_logger.exception("LoggingWorker error: %s", e)
finally:
@ -171,6 +187,35 @@ class LoggingWorker:
# Always release semaphore, even if queue is None
sem.release()
def _record_callback_timeout(self, coroutine: Coroutine) -> None:
"""Count a callback timeout and arm a debounced summary, so a burst of timeouts
(e.g. a slow Redis timing out many callbacks at once) logs one bounded line rather
than a full ERROR stacktrace per callback."""
self._timeout_total += 1
self._timeout_burst_count += 1
self._timeout_last_callback = _coroutine_name(coroutine)
if self._timeout_summary_task is None or self._timeout_summary_task.done():
self._timeout_summary_task = asyncio.create_task(self._flush_timeout_summary())
async def _flush_timeout_summary(self) -> None:
"""After the burst settles, log one bounded summary covering every timeout in it."""
await asyncio.sleep(self.timeout_summary_window)
self._emit_timeout_summary()
def _emit_timeout_summary(self) -> None:
"""Log one bounded summary for the current burst and reset the burst counter."""
burst_count: Final = self._timeout_burst_count
self._timeout_burst_count = 0
if burst_count <= 0:
return
verbose_logger.warning(
"LoggingWorker: %d callback(s) timed out after %ss (callback: %s); %d timed out since start",
burst_count,
self.timeout,
self._timeout_last_callback,
self._timeout_total,
)
async def _worker_loop(self) -> None:
"""Main worker loop that gets tasks and schedules them to run concurrently."""
try:
@ -406,6 +451,11 @@ class LoggingWorker:
async def stop(self) -> None:
"""Stop the logging worker and clean up resources."""
if self._timeout_summary_task is not None:
self._timeout_summary_task.cancel()
self._timeout_summary_task = None
self._emit_timeout_summary()
if self._worker_task is None and not self._running_tasks:
# No worker launched and no in-flight tasks to drain.
return

241
tests/test_litellm/litellm_core_utils/test_logging_worker.py
View file

@ -14,6 +14,18 @@ from litellm.constants import LOGGING_WORKER_AGGRESSIVE_CLEAR_COOLDOWN_SECONDS
from litellm.litellm_core_utils.logging_worker import LoggingWorker
class _RecordCollector(logging.Handler):
"""Captures emitted log records so a test can assert on real logging output
(level, message args, traceback) instead of patching the logger object."""
def __init__(self) -> None:
super().__init__()
self.records: list[logging.LogRecord] = []
def emit(self, record: logging.LogRecord) -> None:
self.records.append(record)
class TestLoggingWorker:
"""Test cases for LoggingWorker functionality."""
@ -26,9 +38,7 @@ class TestLoggingWorker:
async def test_graceful_shutdown_with_clear_queue(self, logging_worker):
"""Test that cancellation triggers clear_queue to prevent 'never awaited' warnings."""
# Mock the clear_queue method to verify it's called during cancellation
with patch.object(
logging_worker, "clear_queue", new_callable=AsyncMock
) as mock_clear_queue:
with patch.object(logging_worker, "clear_queue", new_callable=AsyncMock) as mock_clear_queue:
# Start the worker
logging_worker.start()
@ -195,9 +205,7 @@ class TestLoggingWorker:
async def test_worker_handles_cancellation_gracefully(self, logging_worker):
"""Test that the worker handles cancellation without throwing exceptions."""
# Mock verbose_logger to capture debug messages
with patch(
"litellm.litellm_core_utils.logging_worker.verbose_logger"
) as mock_logger:
with patch("litellm.litellm_core_utils.logging_worker.verbose_logger") as mock_logger:
# Start the worker
logging_worker.start()
@ -264,29 +272,21 @@ class TestLoggingWorker:
small_worker._ensure_queue()
# Mock verbose_logger to capture exception messages
with patch(
"litellm.litellm_core_utils.logging_worker.verbose_logger"
) as mock_logger:
with patch("litellm.litellm_core_utils.logging_worker.verbose_logger") as mock_logger:
# Fill the queue beyond capacity
mock_coro = AsyncMock()
for _ in range(5): # More than max_queue_size of 2
small_worker.enqueue(mock_coro())
# Should have logged queue full exceptions
exception_calls = [
call
for call in mock_logger.exception.call_args_list
if "queue is full" in str(call)
]
exception_calls = [call for call in mock_logger.exception.call_args_list if "queue is full" in str(call)]
assert len(exception_calls) > 0
@pytest.mark.asyncio
async def test_context_propagation(self, logging_worker):
"""Test that enqueued tasks execute in their original context."""
# Create a context variable for testing
test_context_var: contextvars.ContextVar[str] = contextvars.ContextVar(
"test_context_var"
)
test_context_var: contextvars.ContextVar[str] = contextvars.ContextVar("test_context_var")
# Track results from multiple tasks using asyncio.Event for synchronization
task_results = []
@ -364,36 +364,28 @@ class TestLoggingWorker:
task_results.sort(key=lambda x: x["task_id"])
# Verify that each task saw its own context
assert (
len(task_results) == 3
), f"Expected 3 results, got {len(task_results)}: {task_results}"
assert len(task_results) == 3, f"Expected 3 results, got {len(task_results)}: {task_results}"
# Task 1 should see "context_1"
task1_result = next((r for r in task_results if r["task_id"] == "task_1"), None)
assert task1_result is not None, "Task 1 result not found"
assert (
task1_result["context_accessible"] is True
), "Task 1 should have access to context variable"
assert (
task1_result["context_value"] == "context_1"
), f"Task 1 should see 'context_1', got: {task1_result['context_value']}"
assert task1_result["context_accessible"] is True, "Task 1 should have access to context variable"
assert task1_result["context_value"] == "context_1", (
f"Task 1 should see 'context_1', got: {task1_result['context_value']}"
)
# Task 2 should see "context_2"
task2_result = next((r for r in task_results if r["task_id"] == "task_2"), None)
assert task2_result is not None, "Task 2 result not found"
assert (
task2_result["context_accessible"] is True
), "Task 2 should have access to context variable"
assert (
task2_result["context_value"] == "context_2"
), f"Task 2 should see 'context_2', got: {task2_result['context_value']}"
assert task2_result["context_accessible"] is True, "Task 2 should have access to context variable"
assert task2_result["context_value"] == "context_2", (
f"Task 2 should see 'context_2', got: {task2_result['context_value']}"
)
# Task 3 should not have access to the context variable
task3_result = next((r for r in task_results if r["task_id"] == "task_3"), None)
assert task3_result is not None, "Task 3 result not found"
assert (
task3_result["context_accessible"] is False
), "Task 3 should not have access to context variable"
assert task3_result["context_accessible"] is False, "Task 3 should not have access to context variable"
@pytest.mark.asyncio
async def test_semaphore_concurrency_limit(self):
@ -525,3 +517,182 @@ class TestLoggingWorker:
asyncio.run(rebind_on_second_loop())
assert sorted(executed) == [0, 1, 2, 3, 4]
@pytest.mark.asyncio
async def test_timeout_burst_logs_one_bounded_summary(self):
"""Regression (LIT-7519): a burst of callback timeouts must log one bounded summary,
not a full ERROR traceback per timed-out callback.
Before the fix every timed-out callback hit ``verbose_logger.exception`` in
``_process_log_task``, so a slow Redis timing out many spend-tracking callbacks at once
produced one stacktrace each, clustered milliseconds apart. They must collapse into a
single WARNING that counts them, with no tracebacks.
"""
timeout_count = 25
worker = LoggingWorker(
timeout=0.05,
max_queue_size=100,
concurrency=100,
timeout_summary_window=1.0,
)
async def slow_callback() -> None:
await asyncio.sleep(10)
logger = logging.getLogger("LiteLLM")
collector = _RecordCollector()
previous_level = logger.level
logger.addHandler(collector)
logger.setLevel(logging.DEBUG)
try:
worker.start()
for _ in range(timeout_count):
worker.enqueue(slow_callback())
await worker.flush()
assert worker._timeout_summary_task is not None
await worker._timeout_summary_task
await worker.stop()
finally:
logger.removeHandler(collector)
logger.setLevel(previous_level)
errors = [r for r in collector.records if r.levelno >= logging.ERROR]
warnings = [r for r in collector.records if r.levelno == logging.WARNING]
assert errors == [], "a timeout burst must not emit any ERROR tracebacks"
assert len(warnings) == 1, "the whole burst must collapse into one summary line"
summary = warnings[0].getMessage()
assert f"{timeout_count} callback(s) timed out" in summary
assert f"{timeout_count} timed out since start" in summary
assert "slow_callback" in summary
@pytest.mark.asyncio
async def test_non_timeout_error_keeps_traceback(self):
"""A real programming error in a callback must still log a full traceback, so the
timeout aggregation never hides genuine failures.
"""
worker = LoggingWorker(timeout=5.0, max_queue_size=10)
async def failing_callback() -> None:
raise ValueError("boom")
logger = logging.getLogger("LiteLLM")
collector = _RecordCollector()
previous_level = logger.level
logger.addHandler(collector)
logger.setLevel(logging.DEBUG)
try:
worker.start()
worker.enqueue(failing_callback())
await worker.flush()
await worker.stop()
finally:
logger.removeHandler(collector)
logger.setLevel(previous_level)
errors = [r for r in collector.records if r.levelno >= logging.ERROR]
warnings = [r for r in collector.records if r.levelno == logging.WARNING]
assert len(errors) == 1, "a real error must still be logged once"
assert errors[0].exc_info is not None, "the traceback must be preserved"
assert warnings == [], "a single real error is not a timeout summary"
assert worker._timeout_summary_task is None
@pytest.mark.asyncio
async def test_callback_raised_timeout_keeps_traceback(self):
"""A callback that raises TimeoutError on its own did not hit the worker's deadline,
so it is a real failure and must keep its traceback rather than being folded into the
bounded burst summary.
"""
worker = LoggingWorker(timeout=5.0, max_queue_size=10, timeout_summary_window=1.0)
async def raises_own_timeout() -> None:
raise asyncio.TimeoutError("callback's own downstream timeout")
logger = logging.getLogger("LiteLLM")
collector = _RecordCollector()
previous_level = logger.level
logger.addHandler(collector)
logger.setLevel(logging.DEBUG)
try:
worker.start()
worker.enqueue(raises_own_timeout())
await worker.flush()
await worker.stop()
finally:
logger.removeHandler(collector)
logger.setLevel(previous_level)
errors = [r for r in collector.records if r.levelno >= logging.ERROR]
warnings = [r for r in collector.records if r.levelno == logging.WARNING]
assert len(errors) == 1, "a callback-raised TimeoutError must still be logged once"
assert errors[0].exc_info is not None, "the traceback must be preserved"
assert warnings == [], "a callback-raised TimeoutError is not a worker-deadline timeout"
assert worker._timeout_summary_task is None
@pytest.mark.asyncio
async def test_stop_flushes_pending_timeout_summary(self):
"""A burst still inside its summary window when the worker stops must still emit its one
summary, instead of losing it when the event loop tears down.
"""
worker = LoggingWorker(
timeout=0.05,
max_queue_size=10,
concurrency=10,
timeout_summary_window=30.0,
)
async def slow_callback() -> None:
await asyncio.sleep(10)
logger = logging.getLogger("LiteLLM")
collector = _RecordCollector()
previous_level = logger.level
logger.addHandler(collector)
logger.setLevel(logging.DEBUG)
try:
worker.start()
for _ in range(3):
worker.enqueue(slow_callback())
await worker.flush()
assert worker._timeout_summary_task is not None
assert not worker._timeout_summary_task.done(), "precondition: the summary window has not elapsed"
await worker.stop()
finally:
logger.removeHandler(collector)
logger.setLevel(previous_level)
warnings = [r for r in collector.records if r.levelno == logging.WARNING]
assert len(warnings) == 1, "stop must flush the pending summary exactly once"
assert "3 callback(s) timed out" in warnings[0].getMessage()
assert worker._timeout_summary_task is None
def test_loop_change_resets_timeout_summary_state(self):
"""On an event-loop change the summary task is bound to the dead loop; it and the pending
burst count must reset so timeouts on the new loop arm a fresh summary rather than a stale,
stuck one that silently drops later summaries.
"""
worker = LoggingWorker(timeout=1.0, max_queue_size=10, timeout_summary_window=30.0)
async def timed_out_callback() -> None:
await asyncio.sleep(10)
async def arm_on_first_loop() -> None:
worker._ensure_queue()
coro = timed_out_callback()
worker._record_callback_timeout(coro)
coro.close()
assert worker._timeout_summary_task is not None
assert worker._timeout_burst_count == 1
asyncio.run(arm_on_first_loop())
assert worker._timeout_summary_task is not None
async def rebind_on_second_loop() -> None:
worker._ensure_queue()
assert worker._timeout_summary_task is None, "stale summary task must drop on loop change"
assert worker._timeout_burst_count == 0, "stale burst count must reset on loop change"
asyncio.run(rebind_on_second_loop())