mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-06 02:48:13 +00:00
fix(proxy): requeue spend logs on local pre-write failures
This commit is contained in:
parent
74cad08997
commit
5480ddf3d7
2 changed files with 87 additions and 2 deletions
|
|
@ -7288,7 +7288,12 @@ class ProxyUpdateSpend:
|
|||
if not base_url.endswith("/"):
|
||||
base_url += "/"
|
||||
verbose_proxy_logger.debug("base_url: %s", base_url)
|
||||
json_data = json.dumps(logs_to_process)
|
||||
try:
|
||||
json_data = json.dumps(logs_to_process)
|
||||
except (TypeError, ValueError):
|
||||
# No external request has been sent. The batch is safe to replay.
|
||||
await requeue_spend_logs(prisma_client, proxy_logging_obj, logs_to_process)
|
||||
raise
|
||||
response = await db_writer_client.post(
|
||||
url=base_url + "spend/update",
|
||||
data=json_data,
|
||||
|
|
@ -7301,7 +7306,13 @@ class ProxyUpdateSpend:
|
|||
else:
|
||||
for j in range(0, len(logs_to_process), BATCH_SIZE):
|
||||
batch = logs_to_process[j : j + BATCH_SIZE]
|
||||
batch_with_dates = [prisma_client.jsonify_object({**entry}) for entry in batch]
|
||||
try:
|
||||
batch_with_dates = [prisma_client.jsonify_object({**entry}) for entry in batch]
|
||||
except (TypeError, ValueError):
|
||||
# This batch has not reached Prisma. Earlier batches may
|
||||
# already have committed, so replay only this tail.
|
||||
await requeue_spend_logs(prisma_client, proxy_logging_obj, logs_to_process[j:])
|
||||
raise
|
||||
isolation_budget = MAX_SPEND_LOG_ISOLATION_FAILURES_PER_BATCH
|
||||
for statement_rows in spend_log_write_batches(
|
||||
batch_with_dates,
|
||||
|
|
|
|||
|
|
@ -917,3 +917,77 @@ async def test_update_spend_logs_parks_failed_batch_in_redis_with_wire_safe_date
|
|||
parked = await buffer.get_spend_logs_from_redis_buffer(limit=10)
|
||||
assert mock_prisma_client.spend_log_transactions == []
|
||||
assert [(row["request_id"], row["startTime"]) for row in parked] == [("a", started.isoformat())]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_spend_log_serialization_failure_requeues_only_unwritten_tail(
|
||||
mock_prisma_client: Any, make_spend_log_row: Any, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.delenv("SPEND_LOGS_URL", raising=False)
|
||||
rows = [make_spend_log_row(request_id=f"committed-{i}") for i in range(1000)]
|
||||
rows.append(make_spend_log_row(request_id="unwritten"))
|
||||
original_jsonify = mock_prisma_client.jsonify_object
|
||||
|
||||
def jsonify(row: Any) -> Any:
|
||||
if row["request_id"] == "unwritten":
|
||||
raise TypeError("bad local serialization")
|
||||
return original_jsonify(row)
|
||||
|
||||
mock_prisma_client.jsonify_object = jsonify
|
||||
mock_prisma_client.db.litellm_spendlogs.create_many = AsyncMock()
|
||||
proxy_logging = MagicMock()
|
||||
proxy_logging.failure_handler = AsyncMock()
|
||||
with pytest.raises(TypeError, match="bad local serialization"):
|
||||
await ProxyUpdateSpend.update_spend_logs(
|
||||
n_retry_times=0,
|
||||
prisma_client=mock_prisma_client,
|
||||
db_writer_client=None,
|
||||
proxy_logging_obj=proxy_logging,
|
||||
logs_to_process=rows,
|
||||
)
|
||||
assert mock_prisma_client.db.litellm_spendlogs.create_many.await_count >= 1
|
||||
assert [row["request_id"] for row in mock_prisma_client.spend_log_transactions] == ["unwritten"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_external_spend_log_preflight_failure_requeues_without_post(
|
||||
mock_prisma_client: Any, make_spend_log_row: Any, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setenv("SPEND_LOGS_URL", "http://writer.invalid")
|
||||
rows = [make_spend_log_row(request_id="unwritten")]
|
||||
rows[0]["unserializable"] = object()
|
||||
writer = MagicMock()
|
||||
writer.post = AsyncMock()
|
||||
proxy_logging = MagicMock()
|
||||
proxy_logging.failure_handler = AsyncMock()
|
||||
with pytest.raises(TypeError):
|
||||
await ProxyUpdateSpend.update_spend_logs(
|
||||
n_retry_times=0,
|
||||
prisma_client=mock_prisma_client,
|
||||
db_writer_client=writer,
|
||||
proxy_logging_obj=proxy_logging,
|
||||
logs_to_process=rows,
|
||||
)
|
||||
writer.post.assert_not_awaited()
|
||||
assert [row["request_id"] for row in mock_prisma_client.spend_log_transactions] == ["unwritten"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_external_spend_log_post_error_is_not_replayed(
|
||||
mock_prisma_client: Any, make_spend_log_row: Any, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setenv("SPEND_LOGS_URL", "http://writer.invalid")
|
||||
writer = MagicMock()
|
||||
writer.post = AsyncMock(side_effect=ValueError("uncertain delivery"))
|
||||
proxy_logging = MagicMock()
|
||||
proxy_logging.failure_handler = AsyncMock()
|
||||
with pytest.raises(ValueError, match="uncertain delivery"):
|
||||
await ProxyUpdateSpend.update_spend_logs(
|
||||
n_retry_times=0,
|
||||
prisma_client=mock_prisma_client,
|
||||
db_writer_client=writer,
|
||||
proxy_logging_obj=proxy_logging,
|
||||
logs_to_process=[make_spend_log_row(request_id="maybe-committed")],
|
||||
)
|
||||
writer.post.assert_awaited_once()
|
||||
assert mock_prisma_client.spend_log_transactions == []
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue