From 1d039ed0095818beb522c90792796eef5ac28420 Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 25 Sep 2026 09:40:20 -0700 Subject: [PATCH] fix(proxy): give SpendLogToolIndex its own share of the cleanup budget and log a per-run summary (#41768) * fix(proxy): give SpendLogToolIndex its own share of the cleanup budget and log a per-run summary Resolves LIT-8090 starvation bug: _clean_spend_log_tables gave LiteLLM_SpendLogs and LiteLLM_SpendLogToolIndex one shared deadline, so a persistent SpendLogs backlog starved the index table of every delete batch. Split the group deadline between the two tables and emit one per-run summary line (WARNING when backlog remains). Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * ci: rerun unit tests after an order-dependent allowlist flake Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: yassin Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../db_transaction_queue/spend_log_cleanup.py | 27 ++++-- .../proxy/test_spend_log_cleanup.py | 94 ++++++++++++++++++- 2 files changed, 114 insertions(+), 7 deletions(-) diff --git a/litellm/proxy/db/db_transaction_queue/spend_log_cleanup.py b/litellm/proxy/db/db_transaction_queue/spend_log_cleanup.py index 85e19fa8a32..c6f52bf074b 100644 --- a/litellm/proxy/db/db_transaction_queue/spend_log_cleanup.py +++ b/litellm/proxy/db/db_transaction_queue/spend_log_cleanup.py @@ -37,6 +37,7 @@ StopReason: TypeAlias = Literal["exhausted", "budget_exhausted", "batch_cap_reac class TableCleanupResult: """Outcome of pruning one table, so the caller can report why a run ended.""" + table_name: str rows_deleted: int stop_reason: StopReason @@ -472,11 +473,11 @@ class SpendLogCleanup: from the last run that finished inside its budget. """ if time.monotonic() >= deadline: - return TableCleanupResult(rows_deleted=rows_deleted, stop_reason=stop_reason) + return TableCleanupResult(table_name=table_name, rows_deleted=rows_deleted, stop_reason=stop_reason) remaining: Final = await self._count_remaining(prisma_client, cutoff_date, table_name, time_column, deadline) if remaining is not None: SpendLogCleanupMetrics.set_rows_remaining(table_name, remaining) - return TableCleanupResult(rows_deleted=rows_deleted, stop_reason=stop_reason) + return TableCleanupResult(table_name=table_name, rows_deleted=rows_deleted, stop_reason=stop_reason) async def _delete_old_logs( self, prisma_client: PrismaClient, cutoff_date: datetime, deadline: float @@ -571,7 +572,9 @@ class SpendLogCleanup: ) verbose_proxy_logger.info("Dropped %d expired spend-log partitions: %s", len(dropped), dropped) - logs_result: Final = await self._delete_old_logs(prisma_client, cutoff_date, deadline) + logs_result: Final = await self._delete_old_logs( + prisma_client, cutoff_date, self._group_deadline(deadline, groups_remaining=2) + ) verbose_proxy_logger.info("Deleted %s logs", logs_result.rows_deleted) index_result: Final = await self._delete_old_tool_index_rows(prisma_client, cutoff_date, deadline) @@ -638,6 +641,17 @@ class SpendLogCleanup: return "batch_cap_reached" return "completed" + @staticmethod + def _log_run_summary(outcome: RunOutcome, results: tuple[TableCleanupResult, ...], elapsed_seconds: float) -> None: + per_table: Final = ", ".join( + f"{result.table_name}: deleted={result.rows_deleted} stop_reason={result.stop_reason}" for result in results + ) + message: Final = "Spend log cleanup run finished: outcome=%s elapsed=%.1fs [%s]" + if outcome == "completed": + verbose_proxy_logger.info(message, outcome, elapsed_seconds, per_table) + return + verbose_proxy_logger.warning(message, outcome, elapsed_seconds, per_table) + async def cleanup_old_spend_logs(self, prisma_client: PrismaClient) -> None: """ Main cleanup function. Deletes old spend logs in batches. @@ -724,9 +738,10 @@ class SpendLogCleanup: else () ) - SpendLogCleanupMetrics.record_run( - self._run_outcome(spend_log_results + session_results + health_check_results) - ) + results: Final = spend_log_results + session_results + health_check_results + outcome: Final = self._run_outcome(results) + SpendLogCleanupMetrics.record_run(outcome) + self._log_run_summary(outcome, results, time.monotonic() - run_started_at) except asyncio.CancelledError: verbose_proxy_logger.error( diff --git a/tests/test_litellm/proxy/test_spend_log_cleanup.py b/tests/test_litellm/proxy/test_spend_log_cleanup.py index e333da03950..72463e17c6b 100644 --- a/tests/test_litellm/proxy/test_spend_log_cleanup.py +++ b/tests/test_litellm/proxy/test_spend_log_cleanup.py @@ -3,6 +3,7 @@ Test cases for spend log cleanup functionality """ import asyncio +import logging import math import time from contextlib import asynccontextmanager @@ -1421,7 +1422,10 @@ def test_the_reported_run_outcome_is_the_most_significant_reason_in_any_order(st results into one answer: a first-match-wins implementation would pass on whichever order happened to be written and fail on its mirror. """ - results = tuple(TableCleanupResult(rows_deleted=0, stop_reason=reason) for reason in stop_reasons) + results = tuple( + TableCleanupResult(table_name=f"t{i}", rows_deleted=0, stop_reason=reason) + for i, reason in enumerate(stop_reasons) + ) assert SpendLogCleanup._run_outcome(results) == expected @@ -1545,3 +1549,91 @@ async def test_progress_reported_by_an_overlapping_run_is_its_own(monkeypatch): (error_call,) = mock_logger.error.call_args_list rendered = error_call[0][0] % error_call[0][1:] assert "(rows_deleted=100, batches=1)" in rendered + + +@pytest.mark.asyncio +async def test_spend_logs_backlog_cannot_starve_tool_index_cleanup(): + """ + Both spend-log tables share one run budget. Before the fix the spend-log + loop ran against the whole deadline, so a backlog that outlasted the budget + meant LiteLLM_SpendLogToolIndex never received a single delete batch, run + after run. The index table must still get its own share of the budget. + """ + mock_prisma_client = MagicMock() + mock_db = MagicMock() + _wire_tx(mock_db) + mock_db.execute_raw = AsyncMock(return_value=1000) + mock_prisma_client.db = mock_db + + cleaner = SpendLogCleanup( + general_settings={ + "maximum_spend_logs_retention_period": "7d", + "maximum_spend_logs_cleanup_max_batches": 500, + "maximum_spend_logs_cleanup_run_budget": "1s", + } + ) + cleaner.pod_lock_manager = None + + started_at = time.monotonic() + await cleaner.cleanup_old_spend_logs(mock_prisma_client) + elapsed = time.monotonic() - started_at + + tables = [call[0][0].split('"')[1] for call in mock_db.execute_raw.call_args_list] + assert tables.count("LiteLLM_SpendLogs") > 0 + assert tables.count("LiteLLM_SpendLogToolIndex") > 0, "tool index cleanup was starved by the spend-log backlog" + assert elapsed < 2.5, f"splitting the budget must not extend the run: {elapsed}s" + + +@pytest.mark.asyncio +async def test_run_that_leaves_backlog_logs_a_warning_summary_naming_each_table(caplog): + """ + Operators running at warning or error level saw nothing when a run stopped + with expired rows still present. A run that ends on a bound must emit one + WARNING line that names every table, its rows deleted and its stop reason. + """ + mock_prisma_client = MagicMock() + mock_db = MagicMock() + _wire_tx(mock_db) + mock_db.execute_raw = AsyncMock(return_value=1000) + mock_prisma_client.db = mock_db + + cleaner = SpendLogCleanup( + general_settings={ + "maximum_spend_logs_retention_period": "7d", + "maximum_spend_logs_cleanup_max_batches": 2, + } + ) + cleaner.pod_lock_manager = None + + with caplog.at_level(logging.WARNING, logger="LiteLLM Proxy"): + await cleaner.cleanup_old_spend_logs(mock_prisma_client) + + summaries = [record for record in caplog.records if "Spend log cleanup run finished" in record.getMessage()] + assert len(summaries) == 1 + summary = summaries[0] + assert summary.levelno == logging.WARNING + message = summary.getMessage() + assert "outcome=batch_cap_reached" in message + assert "LiteLLM_SpendLogs: deleted=2000 stop_reason=batch_cap_reached" in message + assert "LiteLLM_SpendLogToolIndex: deleted=2000 stop_reason=batch_cap_reached" in message + + +@pytest.mark.asyncio +async def test_run_that_drains_every_table_logs_the_summary_at_info_not_warning(caplog): + """A healthy run must not page anyone: the summary stays at INFO.""" + mock_prisma_client = MagicMock() + mock_db = MagicMock() + _wire_tx(mock_db) + mock_db.execute_raw = AsyncMock(return_value=0) + mock_prisma_client.db = mock_db + + cleaner = SpendLogCleanup(general_settings={"maximum_spend_logs_retention_period": "7d"}) + cleaner.pod_lock_manager = None + + with caplog.at_level(logging.INFO, logger="LiteLLM Proxy"): + await cleaner.cleanup_old_spend_logs(mock_prisma_client) + + summaries = [record for record in caplog.records if "Spend log cleanup run finished" in record.getMessage()] + assert len(summaries) == 1 + assert summaries[0].levelno == logging.INFO + assert "outcome=completed" in summaries[0].getMessage()