mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-29 06:33:28 +00:00
feat(proxy): size-triggered spend log cleanup, oldest rows first (#47450)
Adds general_settings.maximum_spend_logs_size, maximum_spend_logs_cleanup_target_size and maximum_spend_logs_size_check_interval, the spend_log_size_check job, maximum_spend_logs_cleanup_cron precedence, oldest-first bounded deletes down to the target, storage fields in the run summary and the litellm_spend_log_storage_bytes gauge, with a Postgres integration test covering the post-VACUUM no-refire case Resolves LIT-9831 Fixes #44589 Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Kerry Lu
GitHub
parent
4b1928387f
commit
b245cc0d26
8 changed files with 962 additions and 75 deletions
|
|
@ -1855,6 +1855,9 @@ SPEND_LOG_CLEANUP_BATCH_FAILURE_BACKOFF_SECONDS: Final = get_env_float(
|
|||
SPEND_LOG_CLEANUP_RUN_BUDGET_SECONDS: Final = get_env_float("SPEND_LOG_CLEANUP_RUN_BUDGET_SECONDS", 300.0)
|
||||
SPEND_LOG_CLEANUP_BATCH_TIMEOUT_SECONDS: Final = get_env_float("SPEND_LOG_CLEANUP_BATCH_TIMEOUT_SECONDS", 30.0)
|
||||
SPEND_LOG_CLEANUP_REMAINING_COUNT_CAP: Final = get_env_int("SPEND_LOG_CLEANUP_REMAINING_COUNT_CAP", 100000)
|
||||
SPEND_LOG_SIZE_CHECK_JOB_NAME: Final = "spend_log_size_check"
|
||||
SPEND_LOG_SIZE_CHECK_INTERVAL: Final = os.getenv("SPEND_LOG_SIZE_CHECK_INTERVAL", "1d")
|
||||
SPEND_LOG_SIZE_CLEANUP_RECHECK_BATCHES: Final = get_env_int("SPEND_LOG_SIZE_CLEANUP_RECHECK_BATCHES", 10)
|
||||
SCHEDULED_JOB_SHUTDOWN_FINISH_TIMEOUT_SECONDS: Final = get_env_float(
|
||||
"SCHEDULED_JOB_SHUTDOWN_FINISH_TIMEOUT_SECONDS", 5.0
|
||||
)
|
||||
|
|
|
|||
|
|
@ -2884,6 +2884,30 @@ class ConfigGeneralSettings(LiteLLMPydanticObjectBase):
|
|||
"deleted. Only historical tag usage analytics are affected; tag budgets read the lifetime counter."
|
||||
),
|
||||
)
|
||||
maximum_spend_logs_size: str | None = Field(
|
||||
None,
|
||||
description=(
|
||||
"Size rule for spend logs (e.g. '180GB'): when LiteLLM_SpendLogs plus LiteLLM_SpendLogToolIndex "
|
||||
"(heap, indexes and TOAST, measured with pg_total_relation_size) hold more live data than this, the "
|
||||
"cleanup job deletes the oldest spend logs first until they are under "
|
||||
"maximum_spend_logs_cleanup_target_size. Independent of maximum_spend_logs_retention_period: a row is "
|
||||
"deleted when either rule says so. Unset means no size rule."
|
||||
),
|
||||
)
|
||||
maximum_spend_logs_cleanup_target_size: str | None = Field(
|
||||
None,
|
||||
description=(
|
||||
"Level a size-triggered cleanup run deletes down to (e.g. '20GB'). Must be below maximum_spend_logs_size. "
|
||||
"Unset means the run stops as soon as the tables are back under maximum_spend_logs_size."
|
||||
),
|
||||
)
|
||||
maximum_spend_logs_size_check_interval: str | None = Field(
|
||||
None,
|
||||
description=(
|
||||
"How often the size rule is checked (e.g. '10m'). Defaults to '1d'. Ignored when "
|
||||
"maximum_spend_logs_cleanup_cron is set, which then drives both the age and the size cleanup."
|
||||
),
|
||||
)
|
||||
use_spend_logs_partitioning: bool | None = Field(
|
||||
None,
|
||||
description="If True and LiteLLM_SpendLogs has been converted to a range-partitioned table (db_scripts/partition_spend_logs.sql), retention cleanup drops expired partitions instead of deleting rows, and pre-creates upcoming partitions. Default is False.",
|
||||
|
|
|
|||
|
|
@ -1,6 +1,8 @@
|
|||
import asyncio
|
||||
import math
|
||||
import re
|
||||
import time
|
||||
from collections.abc import Callable
|
||||
from contextvars import ContextVar
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
|
@ -19,6 +21,7 @@ from litellm.constants import (
|
|||
SPEND_LOG_CLEANUP_REMAINING_COUNT_CAP,
|
||||
SPEND_LOG_CLEANUP_RUN_BUDGET_SECONDS,
|
||||
SPEND_LOG_RUN_LOOPS,
|
||||
SPEND_LOG_SIZE_CLEANUP_RECHECK_BATCHES,
|
||||
)
|
||||
from litellm.litellm_core_utils.duration_parser import duration_in_seconds
|
||||
from litellm.proxy.db.db_span import db_span
|
||||
|
|
@ -35,7 +38,7 @@ from litellm.proxy.db.db_transaction_queue.spend_logs_partition_manager import (
|
|||
from litellm.proxy.utils import PrismaClient
|
||||
from litellm.types.llms.base import LiteLLMBaseModel
|
||||
|
||||
StopReason: TypeAlias = Literal["exhausted", "budget_exhausted", "batch_cap_reached", "aborted"]
|
||||
StopReason: TypeAlias = Literal["exhausted", "target_reached", "budget_exhausted", "batch_cap_reached", "aborted"]
|
||||
|
||||
_SIZE_RE: Final = re.compile(r"^\s*(\d+(?:\.\d+)?)\s*([KMGT]?)I?B?\s*$", re.IGNORECASE)
|
||||
_SIZE_UNITS: Final = {"": 1, "K": 1024, "M": 1024**2, "G": 1024**3, "T": 1024**4}
|
||||
|
|
@ -61,6 +64,14 @@ def size_in_bytes(value: str | int) -> int:
|
|||
return size
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class SizeRule:
|
||||
"""The size cleanup rule as configured: start cleaning above `limit_bytes`, stop under `target_bytes`."""
|
||||
|
||||
limit_bytes: int
|
||||
target_bytes: int
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class StorageProbe:
|
||||
"""
|
||||
|
|
@ -218,6 +229,42 @@ class TableCleanupResult:
|
|||
stop_reason: StopReason
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _SizeStepState:
|
||||
"""Progress of the size pass across its re-check steps."""
|
||||
|
||||
logs_deleted: int
|
||||
index_deleted: int
|
||||
batches_left: int
|
||||
stop_reason: StopReason | None
|
||||
|
||||
def stopped(self, reason: StopReason) -> "_SizeStepState":
|
||||
return _SizeStepState(
|
||||
logs_deleted=self.logs_deleted,
|
||||
index_deleted=self.index_deleted,
|
||||
batches_left=self.batches_left,
|
||||
stop_reason=reason,
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class SizePassReport:
|
||||
"""What one size pass measured and achieved, for the run summary and the storage gauges."""
|
||||
|
||||
rule: SizeRule
|
||||
before: SpendLogStorage
|
||||
after: SpendLogStorage
|
||||
results: tuple[TableCleanupResult, ...]
|
||||
|
||||
@property
|
||||
def target_reached(self) -> bool:
|
||||
return self.after.estimated_live_bytes <= self.rule.target_bytes
|
||||
|
||||
@property
|
||||
def above_target_bytes(self) -> int:
|
||||
return max(0, self.after.estimated_live_bytes - self.rule.target_bytes)
|
||||
|
||||
|
||||
class _RunProgress:
|
||||
"""How far one cleanup run has got, reported if that run is cancelled"""
|
||||
|
||||
|
|
@ -277,6 +324,13 @@ _SAVE_SIZE_STATE_SQL: Final = """
|
|||
"""
|
||||
|
||||
|
||||
class _OffsetRow(LiteLLMBaseModel):
|
||||
model_config = ConfigDict(frozen=True)
|
||||
startTime: datetime
|
||||
|
||||
|
||||
_OFFSET_ROWS: Final = TypeAdapter(list[_OffsetRow])
|
||||
|
||||
_SPEND_LOG_TABLES: Final = ("LiteLLM_SpendLogs", "LiteLLM_SpendLogToolIndex")
|
||||
|
||||
# Partition children are counted with their parent: a partitioned parent has no storage of its own.
|
||||
|
|
@ -456,6 +510,53 @@ class SpendLogCleanup:
|
|||
self.retention_seconds = self._retention_seconds_for("maximum_spend_logs_retention_period")
|
||||
return self.retention_seconds is not None
|
||||
|
||||
def _setting(self, setting_name: str) -> object:
|
||||
return self.general_settings.get(setting_name) # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] # untyped settings dict
|
||||
|
||||
def _size_bytes_setting(self, setting_name: str) -> int | None:
|
||||
"""Parse one size setting into bytes, or None when unset or unusable."""
|
||||
raw: Final = self._setting(setting_name)
|
||||
if raw is None:
|
||||
return None
|
||||
if not isinstance(raw, (str, int)):
|
||||
verbose_proxy_logger.warning("Invalid %s value: %r, ignoring it", setting_name, raw)
|
||||
return None
|
||||
try:
|
||||
return size_in_bytes(raw)
|
||||
except ValueError as e:
|
||||
verbose_proxy_logger.warning("Invalid %s value: %r (%s), ignoring it", setting_name, raw, e)
|
||||
return None
|
||||
|
||||
def _size_rule(self) -> SizeRule | None:
|
||||
"""
|
||||
The size rule from settings, or None when maximum_spend_logs_size is unset.
|
||||
|
||||
A target at or above the limit cannot be what the operator meant, so it is reported and the run
|
||||
cleans only until it is back under the limit, which is also what an unset target means.
|
||||
"""
|
||||
limit_bytes: Final = self._size_bytes_setting("maximum_spend_logs_size")
|
||||
if limit_bytes is None:
|
||||
return None
|
||||
target_bytes: Final = self._size_bytes_setting("maximum_spend_logs_cleanup_target_size")
|
||||
if target_bytes is None:
|
||||
return SizeRule(limit_bytes=limit_bytes, target_bytes=limit_bytes)
|
||||
if target_bytes >= limit_bytes:
|
||||
verbose_proxy_logger.error(
|
||||
"maximum_spend_logs_cleanup_target_size (%s bytes) must be below maximum_spend_logs_size (%s bytes); "
|
||||
"cleaning down to maximum_spend_logs_size instead",
|
||||
target_bytes,
|
||||
limit_bytes,
|
||||
)
|
||||
return SizeRule(limit_bytes=limit_bytes, target_bytes=limit_bytes)
|
||||
return SizeRule(limit_bytes=limit_bytes, target_bytes=target_bytes)
|
||||
|
||||
def _size_pass_runs_with_age_pass(self) -> bool:
|
||||
"""
|
||||
With maximum_spend_logs_cleanup_cron set there is one schedule for both rules, so the age run carries
|
||||
the size pass; otherwise the size pass has its own job on maximum_spend_logs_size_check_interval.
|
||||
"""
|
||||
return bool(self._setting("maximum_spend_logs_cleanup_cron"))
|
||||
|
||||
async def _probe_storage(self, prisma_client: PrismaClient, deadline: float) -> StorageProbe | None:
|
||||
"""Read what the spend-log tables weigh from the catalog, or None when the probe fails or has no budget."""
|
||||
if time.monotonic() >= deadline:
|
||||
|
|
@ -508,6 +609,32 @@ class SpendLogCleanup:
|
|||
except Exception as e: # noqa: BLE001 - a lost save costs one re-seed from the catalog, it must not fail the run
|
||||
verbose_proxy_logger.warning("Could not save the spend-log size cleanup state: %s: %s", type(e).__name__, e)
|
||||
|
||||
async def _size_step_cutoff(self, prisma_client: PrismaClient, deadline: float, rows: int) -> datetime | None:
|
||||
"""
|
||||
The startTime that puts the `rows` oldest rows behind it.
|
||||
|
||||
Expressing a size step as a cutoff lets the size pass reuse the retention DELETE unchanged, including
|
||||
the tool-index rows that expire with their spend logs. When fewer rows than that remain, the step is
|
||||
everything up to now.
|
||||
"""
|
||||
if time.monotonic() >= deadline:
|
||||
return None
|
||||
timeout_ms: Final = self._timeout_ms(deadline)
|
||||
async with db_span("size_step_cutoff", "LiteLLM_SpendLogs"), bounded_tx(prisma_client, timeout_ms) as tx:
|
||||
await tx.execute_raw(trusted_sql(f"SET LOCAL statement_timeout = {timeout_ms}"))
|
||||
found: Final = _OFFSET_ROWS.validate_python(
|
||||
await tx.query_raw(
|
||||
trusted_sql(
|
||||
'SELECT "startTime" FROM "LiteLLM_SpendLogs" ORDER BY "startTime" ASC OFFSET $1 LIMIT 1'
|
||||
),
|
||||
rows,
|
||||
)
|
||||
)
|
||||
if not found:
|
||||
return datetime.now(timezone.utc)
|
||||
start_time: Final = found[0].startTime
|
||||
return start_time if start_time.tzinfo is not None else start_time.replace(tzinfo=timezone.utc)
|
||||
|
||||
def _timeout_ms(self, deadline: float) -> int:
|
||||
"""
|
||||
The per-statement bound in milliseconds: the batch timeout, or whatever
|
||||
|
|
@ -621,14 +748,17 @@ class SpendLogCleanup:
|
|||
key_columns: tuple[str, ...],
|
||||
time_column: str,
|
||||
deadline: float,
|
||||
batch_cap: int | None = None,
|
||||
) -> TableCleanupResult:
|
||||
"""
|
||||
Delete a table's rows older than the cutoff in batches.
|
||||
|
||||
Stops at whichever bound is reached first: the backlog running out, the
|
||||
shared wall-clock deadline, the per-table batch cap, or too many
|
||||
consecutive batch failures.
|
||||
shared wall-clock deadline, the per-table batch cap (`batch_cap` when a
|
||||
caller spreads one table's cap over several calls, else the configured
|
||||
one), or too many consecutive batch failures.
|
||||
"""
|
||||
max_batches: Final = self.max_batches if batch_cap is None else batch_cap
|
||||
key_list: Final = ", ".join(f'"{col}"' for col in key_columns)
|
||||
delete_sql: Final = f"""
|
||||
DELETE FROM "{table_name}"
|
||||
|
|
@ -651,7 +781,7 @@ class SpendLogCleanup:
|
|||
return await self._finish_table(
|
||||
prisma_client, cutoff_date, table_name, time_column, total_deleted, "budget_exhausted", deadline
|
||||
)
|
||||
if run_count >= self.max_batches:
|
||||
if run_count >= max_batches:
|
||||
verbose_proxy_logger.info(
|
||||
"Max batches reached for %s cleanup, remaining rows will be deleted in next run", table_name
|
||||
)
|
||||
|
|
@ -891,6 +1021,163 @@ class SpendLogCleanup:
|
|||
verbose_proxy_logger.info("Deleted %s expired tool index rows", index_result.rows_deleted)
|
||||
return (logs_result, index_result)
|
||||
|
||||
async def _clean_spend_logs_by_size(
|
||||
self, prisma_client: PrismaClient, rule: SizeRule, deadline: float
|
||||
) -> SizePassReport | None:
|
||||
"""
|
||||
Bring the spend-log tables under the size target, oldest rows first.
|
||||
|
||||
Live bytes are the files on disk less the reclaimable count carried in LiteLLM_Config (see
|
||||
`SizeCleanupState`), rolled forward from the catalog's cumulative insert and delete counters. Rows go in steps: the rows still needed to reach the target follow
|
||||
from bytes per row, and each step turns that many of the oldest rows (at most
|
||||
SPEND_LOG_SIZE_CLEANUP_RECHECK_BATCHES batches) into a cutoff and runs the same bounded DELETE the age
|
||||
rule uses, for the spend logs and the tool index rows derived from them. Progress is counted from the
|
||||
deletes themselves, not re-read from pg_stat_user_tables, whose counters trail a DELETE by up to a
|
||||
second. The run's batch cap and budget apply to the whole pass, so a pass that stops short reports how
|
||||
far above the target it still is and the next check continues from there.
|
||||
"""
|
||||
probe: Final = await self._probe_storage(prisma_client, deadline)
|
||||
if probe is None:
|
||||
return None
|
||||
if probe.stats_are_stale:
|
||||
verbose_proxy_logger.warning(
|
||||
"Spend-log row statistics are stale (allocated=%s bytes, live_rows=0, dead_rows=0); skipping the "
|
||||
'size pass until pg_stat_user_tables is repopulated, ANALYZE "LiteLLM_SpendLogs" refreshes it',
|
||||
probe.allocated_bytes,
|
||||
)
|
||||
return None
|
||||
state: Final = advance_size_state(await self._load_size_state(prisma_client, deadline), probe)
|
||||
before: Final = SpendLogStorage(
|
||||
allocated_bytes=probe.allocated_bytes,
|
||||
reclaimable_bytes=state.reclaimable_bytes,
|
||||
bytes_per_row=state.bytes_per_row,
|
||||
live_rows=probe.live_rows,
|
||||
)
|
||||
SpendLogCleanupMetrics.set_storage_bytes("allocated", before.allocated_bytes)
|
||||
SpendLogCleanupMetrics.set_storage_bytes("estimated_live", before.estimated_live_bytes)
|
||||
if before.estimated_live_bytes <= rule.limit_bytes:
|
||||
verbose_proxy_logger.info(
|
||||
"Spend-log storage is under maximum_spend_logs_size (estimated_live=%s allocated=%s reclaimable=%s "
|
||||
"limit=%s bytes)",
|
||||
before.estimated_live_bytes,
|
||||
before.allocated_bytes,
|
||||
before.reclaimable_bytes,
|
||||
rule.limit_bytes,
|
||||
)
|
||||
SpendLogCleanupMetrics.set_storage_bytes("above_target", 0)
|
||||
await self._save_size_state(prisma_client, deadline, state)
|
||||
return SizePassReport(rule=rule, before=before, after=before, results=())
|
||||
verbose_proxy_logger.warning(
|
||||
"Spend-log storage is over maximum_spend_logs_size (estimated_live=%s allocated=%s reclaimable=%s "
|
||||
"limit=%s bytes); deleting the oldest spend logs down to %s bytes",
|
||||
before.estimated_live_bytes,
|
||||
before.allocated_bytes,
|
||||
before.reclaimable_bytes,
|
||||
rule.limit_bytes,
|
||||
rule.target_bytes,
|
||||
)
|
||||
rows_to_delete: Final = max(
|
||||
0, math.ceil((before.estimated_live_bytes - rule.target_bytes) / before.bytes_per_row)
|
||||
)
|
||||
|
||||
def rows_still_needed(rows_deleted: int) -> int:
|
||||
return rows_to_delete - rows_deleted
|
||||
|
||||
results: Final = await self._delete_oldest_rows_for_size(prisma_client, deadline, rows_still_needed)
|
||||
rows_deleted: Final = sum(result.rows_deleted for result in results)
|
||||
remeasured: Final = await self._probe_storage(prisma_client, deadline)
|
||||
after: Final = before.after_deleting(
|
||||
rows_deleted, remeasured.allocated_bytes if remeasured is not None else before.allocated_bytes
|
||||
)
|
||||
await self._save_size_state(
|
||||
prisma_client,
|
||||
deadline,
|
||||
after.to_state(inserted_total=probe.inserted_total, deleted_total=probe.deleted_total + rows_deleted),
|
||||
)
|
||||
report: Final = SizePassReport(rule=rule, before=before, after=after, results=results)
|
||||
SpendLogCleanupMetrics.set_storage_bytes("allocated", after.allocated_bytes)
|
||||
SpendLogCleanupMetrics.set_storage_bytes("estimated_live", after.estimated_live_bytes)
|
||||
SpendLogCleanupMetrics.set_storage_bytes("above_target", report.above_target_bytes)
|
||||
return report
|
||||
|
||||
async def _delete_oldest_rows_for_size(
|
||||
self,
|
||||
prisma_client: PrismaClient,
|
||||
deadline: float,
|
||||
rows_still_needed: Callable[[int], int],
|
||||
) -> tuple[TableCleanupResult, ...]:
|
||||
state = _SizeStepState( # rebind-ok: one value threaded through the bounded step loop
|
||||
logs_deleted=0, index_deleted=0, batches_left=self.max_batches, stop_reason=None
|
||||
)
|
||||
for _ in range(self.max_batches // SPEND_LOG_SIZE_CLEANUP_RECHECK_BATCHES + 2):
|
||||
if state.stop_reason is not None:
|
||||
break
|
||||
state = await self._run_size_step(prisma_client, deadline, rows_still_needed, state)
|
||||
final_reason: Final[StopReason] = state.stop_reason or "batch_cap_reached"
|
||||
return (
|
||||
TableCleanupResult(
|
||||
table_name="LiteLLM_SpendLogs (size rule)", rows_deleted=state.logs_deleted, stop_reason=final_reason
|
||||
),
|
||||
TableCleanupResult(
|
||||
table_name="LiteLLM_SpendLogToolIndex (size rule)",
|
||||
rows_deleted=state.index_deleted,
|
||||
stop_reason=final_reason,
|
||||
),
|
||||
)
|
||||
|
||||
async def _run_size_step(
|
||||
self,
|
||||
prisma_client: PrismaClient,
|
||||
deadline: float,
|
||||
rows_still_needed: Callable[[int], int],
|
||||
state: "_SizeStepState",
|
||||
) -> "_SizeStepState":
|
||||
"""One step: delete the rows still needed, at most SPEND_LOG_SIZE_CLEANUP_RECHECK_BATCHES batches of the oldest."""
|
||||
needed: Final = rows_still_needed(state.logs_deleted + state.index_deleted)
|
||||
if needed <= 0:
|
||||
return state.stopped("target_reached")
|
||||
if time.monotonic() >= deadline:
|
||||
return state.stopped("budget_exhausted")
|
||||
if state.batches_left <= 0:
|
||||
return state.stopped("batch_cap_reached")
|
||||
step_cap: Final = min(
|
||||
SPEND_LOG_SIZE_CLEANUP_RECHECK_BATCHES, state.batches_left, math.ceil(needed / self.batch_size)
|
||||
)
|
||||
cutoff: Final = await self._size_step_cutoff(prisma_client, deadline, step_cap * self.batch_size)
|
||||
if cutoff is None:
|
||||
return state.stopped("budget_exhausted")
|
||||
logs_result: Final = await self._delete_old_rows_batched(
|
||||
prisma_client,
|
||||
cutoff,
|
||||
table_name="LiteLLM_SpendLogs",
|
||||
key_columns=("request_id", "startTime"),
|
||||
time_column="startTime",
|
||||
deadline=deadline,
|
||||
batch_cap=step_cap,
|
||||
)
|
||||
index_result: Final = await self._delete_old_rows_batched(
|
||||
prisma_client,
|
||||
cutoff,
|
||||
table_name="LiteLLM_SpendLogToolIndex",
|
||||
key_columns=("request_id", "tool_name"),
|
||||
time_column="start_time",
|
||||
deadline=deadline,
|
||||
batch_cap=step_cap,
|
||||
)
|
||||
advanced: Final = _SizeStepState(
|
||||
logs_deleted=state.logs_deleted + logs_result.rows_deleted,
|
||||
index_deleted=state.index_deleted + index_result.rows_deleted,
|
||||
batches_left=state.batches_left - step_cap,
|
||||
stop_reason=None,
|
||||
)
|
||||
if rows_still_needed(advanced.logs_deleted + advanced.index_deleted) <= 0:
|
||||
return advanced.stopped("target_reached")
|
||||
if logs_result.stop_reason in ("aborted", "budget_exhausted") or index_result.stop_reason == "aborted":
|
||||
return advanced.stopped(logs_result.stop_reason if logs_result.stop_reason != "exhausted" else "aborted")
|
||||
if logs_result.rows_deleted == 0:
|
||||
return advanced.stopped("exhausted")
|
||||
return advanced
|
||||
|
||||
async def _clean_session_rollup(
|
||||
self, prisma_client: PrismaClient, retention_seconds: int, deadline: float
|
||||
) -> tuple[TableCleanupResult, ...]:
|
||||
|
|
@ -968,22 +1255,60 @@ class SpendLogCleanup:
|
|||
return "completed"
|
||||
|
||||
@staticmethod
|
||||
def _log_run_summary(outcome: RunOutcome, results: tuple[TableCleanupResult, ...], elapsed_seconds: float) -> None:
|
||||
def _storage_summary(report: SizePassReport | None) -> str:
|
||||
"""
|
||||
The size pass in one clause: what was measured, where it ended against the target, and whether the
|
||||
operator still has to act to get the disk back.
|
||||
"""
|
||||
if report is None:
|
||||
return ""
|
||||
rows_deleted: Final = sum(result.rows_deleted for result in report.results)
|
||||
reclaim_note: Final = (
|
||||
"; allocated bytes stay until autovacuum reuses them, only VACUUM FULL or pg_repack on the spend-log "
|
||||
"tables returns them to the filesystem"
|
||||
if rows_deleted > 0 and report.after.allocated_bytes >= report.before.allocated_bytes
|
||||
else ""
|
||||
)
|
||||
return (
|
||||
f" storage: allocated_before={report.before.allocated_bytes} "
|
||||
f"estimated_live_before={report.before.estimated_live_bytes} "
|
||||
f"allocated_after={report.after.allocated_bytes} estimated_live_after={report.after.estimated_live_bytes} "
|
||||
f"reclaimable_after={report.after.reclaimable_bytes} "
|
||||
f"limit={report.rule.limit_bytes} target={report.rule.target_bytes} "
|
||||
f"target_reached={str(report.target_reached).lower()} above_target={report.above_target_bytes}{reclaim_note}"
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _log_run_summary(
|
||||
cls,
|
||||
outcome: RunOutcome,
|
||||
results: tuple[TableCleanupResult, ...],
|
||||
elapsed_seconds: float,
|
||||
size_report: SizePassReport | None = None,
|
||||
) -> 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)
|
||||
message: Final = "Spend log cleanup run finished: outcome=%s elapsed=%.1fs [%s]%s"
|
||||
storage: Final = cls._storage_summary(size_report)
|
||||
if outcome == "completed" and (size_report is None or size_report.target_reached):
|
||||
verbose_proxy_logger.info(message, outcome, elapsed_seconds, per_table, storage)
|
||||
return
|
||||
verbose_proxy_logger.warning(message, outcome, elapsed_seconds, per_table)
|
||||
verbose_proxy_logger.warning(message, outcome, elapsed_seconds, per_table, storage)
|
||||
|
||||
async def cleanup_old_spend_logs(self, prisma_client: PrismaClient) -> None:
|
||||
"""
|
||||
Main cleanup function. Deletes old spend logs in batches.
|
||||
The scheduled retention run: every age rule, plus the size rule when one cron drives both.
|
||||
If pod_lock_manager is available, ensures only one pod runs cleanup.
|
||||
If no pod_lock_manager, runs cleanup without distributed locking.
|
||||
"""
|
||||
await self._run(prisma_client, age_rules=True, size_rule=self._size_pass_runs_with_age_pass())
|
||||
|
||||
async def cleanup_spend_logs_by_size(self, prisma_client: PrismaClient) -> None:
|
||||
"""The size-check run scheduled on maximum_spend_logs_size_check_interval: the size rule only."""
|
||||
await self._run(prisma_client, age_rules=False, size_rule=True)
|
||||
|
||||
async def _run(self, prisma_client: PrismaClient, age_rules: bool, size_rule: bool) -> None:
|
||||
lock_acquired = False
|
||||
run_started_at: Final = time.monotonic()
|
||||
progress: Final = _RunProgress()
|
||||
|
|
@ -992,19 +1317,23 @@ class SpendLogCleanup:
|
|||
verbose_proxy_logger.info("Cleanup job triggered at %s", datetime.now())
|
||||
self._refresh_bounds()
|
||||
|
||||
delete_spend_logs: Final = self._should_delete_spend_logs()
|
||||
autorouter_retention_seconds: Final = self._retention_seconds_for(
|
||||
"maximum_autorouter_session_retention_period"
|
||||
delete_spend_logs: Final = age_rules and self._should_delete_spend_logs()
|
||||
active_size_rule: Final = self._size_rule() if size_rule else None
|
||||
autorouter_retention_seconds: Final = (
|
||||
self._retention_seconds_for("maximum_autorouter_session_retention_period") if age_rules else None
|
||||
)
|
||||
health_check_retention_seconds: Final = self._retention_seconds_for("maximum_health_check_retention_period")
|
||||
daily_tag_spend_retention_seconds: Final = self._retention_seconds_for(
|
||||
"maximum_daily_tag_spend_retention_period"
|
||||
health_check_retention_seconds: Final = (
|
||||
self._retention_seconds_for("maximum_health_check_retention_period") if age_rules else None
|
||||
)
|
||||
daily_tag_spend_retention_seconds: Final = (
|
||||
self._retention_seconds_for("maximum_daily_tag_spend_retention_period") if age_rules else None
|
||||
)
|
||||
if (
|
||||
not delete_spend_logs
|
||||
and autorouter_retention_seconds is None
|
||||
and health_check_retention_seconds is None
|
||||
and daily_tag_spend_retention_seconds is None
|
||||
and active_size_rule is None
|
||||
):
|
||||
SpendLogCleanupMetrics.record_run("skipped_disabled")
|
||||
return
|
||||
|
|
@ -1032,11 +1361,13 @@ class SpendLogCleanup:
|
|||
return
|
||||
|
||||
deadline: Final = time.monotonic() + self.run_budget_seconds
|
||||
size_group: Final = int(active_size_rule is not None)
|
||||
configured_group_count: Final = (
|
||||
int(delete_spend_logs and self.retention_seconds is not None)
|
||||
+ int(autorouter_retention_seconds is not None)
|
||||
+ int(health_check_retention_seconds is not None)
|
||||
+ int(daily_tag_spend_retention_seconds is not None)
|
||||
+ size_group
|
||||
)
|
||||
|
||||
spend_log_results: Final = (
|
||||
|
|
@ -1051,9 +1382,12 @@ class SpendLogCleanup:
|
|||
int(autorouter_retention_seconds is not None)
|
||||
+ int(health_check_retention_seconds is not None)
|
||||
+ int(daily_tag_spend_retention_seconds is not None)
|
||||
+ size_group
|
||||
)
|
||||
remaining_groups_after_sessions: Final = int(health_check_retention_seconds is not None) + int(
|
||||
daily_tag_spend_retention_seconds is not None
|
||||
remaining_groups_after_sessions: Final = (
|
||||
int(health_check_retention_seconds is not None)
|
||||
+ int(daily_tag_spend_retention_seconds is not None)
|
||||
+ size_group
|
||||
)
|
||||
session_results: Final = (
|
||||
await self._clean_session_rollup(
|
||||
|
|
@ -1074,15 +1408,27 @@ class SpendLogCleanup:
|
|||
else ()
|
||||
)
|
||||
daily_tag_spend_results: Final = (
|
||||
await self._clean_daily_tag_spend(prisma_client, daily_tag_spend_retention_seconds, deadline)
|
||||
await self._clean_daily_tag_spend(
|
||||
prisma_client,
|
||||
daily_tag_spend_retention_seconds,
|
||||
self._group_deadline(deadline, 1 + size_group),
|
||||
)
|
||||
if daily_tag_spend_retention_seconds is not None
|
||||
else ()
|
||||
)
|
||||
size_report: Final = (
|
||||
await self._clean_spend_logs_by_size(prisma_client, active_size_rule, deadline)
|
||||
if active_size_rule is not None
|
||||
else None
|
||||
)
|
||||
size_results: Final = size_report.results if size_report is not None else ()
|
||||
|
||||
results: Final = spend_log_results + session_results + health_check_results + daily_tag_spend_results
|
||||
results: Final = (
|
||||
spend_log_results + session_results + health_check_results + daily_tag_spend_results + size_results
|
||||
)
|
||||
outcome: Final = self._run_outcome(results)
|
||||
SpendLogCleanupMetrics.record_run(outcome)
|
||||
self._log_run_summary(outcome, results, time.monotonic() - run_started_at)
|
||||
self._log_run_summary(outcome, results, time.monotonic() - run_started_at, size_report)
|
||||
|
||||
except asyncio.CancelledError:
|
||||
verbose_proxy_logger.error(
|
||||
|
|
|
|||
|
|
@ -32,6 +32,9 @@ RunOutcome: TypeAlias = Literal[
|
|||
_BATCH_DURATION_BUCKETS: Final = (0.005, 0.025, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0)
|
||||
_TABLE_LABEL: Final = ("table",)
|
||||
_OUTCOME_LABEL: Final = ("outcome",)
|
||||
_MEASURE_LABEL: Final = ("measure",)
|
||||
|
||||
StorageMeasure: TypeAlias = Literal["allocated", "estimated_live", "above_target"]
|
||||
|
||||
|
||||
class SpendLogCleanupMetrics:
|
||||
|
|
@ -49,6 +52,7 @@ class SpendLogCleanupMetrics:
|
|||
rows_remaining: "PrometheusGauge | None" = None
|
||||
batch_failures: "PrometheusCounter | None" = None
|
||||
runs: "PrometheusCounter | None" = None
|
||||
storage_bytes: "PrometheusGauge | None" = None
|
||||
|
||||
@classmethod
|
||||
def _ensure_initialized(cls) -> None:
|
||||
|
|
@ -90,6 +94,15 @@ class SpendLogCleanupMetrics:
|
|||
"Retention cleanup runs, labelled by why the run ended",
|
||||
labelnames=_OUTCOME_LABEL,
|
||||
)
|
||||
cls.storage_bytes = Gauge(
|
||||
"litellm_spend_log_storage_bytes",
|
||||
"Disk held by LiteLLM_SpendLogs plus LiteLLM_SpendLogToolIndex as last measured by the cleanup "
|
||||
"job: allocated is pg_total_relation_size (heap, indexes, TOAST), estimated_live scales it by the "
|
||||
"live share of rows from pg_stat_user_tables, above_target is estimated_live minus the size "
|
||||
"cleanup target (0 once the target is reached)",
|
||||
labelnames=_MEASURE_LABEL,
|
||||
multiprocess_mode="livemax",
|
||||
)
|
||||
except Exception as e: # noqa: BLE001 - a metrics problem must never fail the cleanup run
|
||||
# Covers the extra being absent, a duplicate registration (repeated
|
||||
# imports under a test runner), and registry misconfiguration alike.
|
||||
|
|
@ -120,3 +133,9 @@ class SpendLogCleanupMetrics:
|
|||
cls._ensure_initialized()
|
||||
if cls.runs is not None:
|
||||
cls.runs.labels(outcome=outcome).inc()
|
||||
|
||||
@classmethod
|
||||
def set_storage_bytes(cls, measure: StorageMeasure, value: int) -> None:
|
||||
cls._ensure_initialized()
|
||||
if cls.storage_bytes is not None:
|
||||
cls.storage_bytes.labels(measure=measure).set(value)
|
||||
|
|
|
|||
|
|
@ -340,6 +340,7 @@ from litellm.constants import (
|
|||
REALTIME_SESSION_SUCCESS_LOGGED_KEY,
|
||||
ROUTER_SETTINGS_MANAGED_OUTSIDE_CONFIG,
|
||||
SPEND_CAPTURE_RATE_CHECK_JOB_ID,
|
||||
SPEND_LOG_SIZE_CHECK_INTERVAL,
|
||||
USER_SPEND_ALERTS_JOB_ID,
|
||||
WEEKLY_SPEND_REPORT_JOB_ID,
|
||||
)
|
||||
|
|
@ -5522,20 +5523,44 @@ def _current_general_settings() -> Mapping[str, object]:
|
|||
return general_settings
|
||||
|
||||
|
||||
_CLEANUP_SCHEDULE_KEYS: Final = (
|
||||
_RETENTION_RULE_KEYS: Final = (
|
||||
"maximum_spend_logs_retention_period",
|
||||
"maximum_autorouter_session_retention_period",
|
||||
"maximum_health_check_retention_period",
|
||||
"maximum_daily_tag_spend_retention_period",
|
||||
)
|
||||
_CLEANUP_SCHEDULE_KEYS: Final = (
|
||||
*_RETENTION_RULE_KEYS,
|
||||
"maximum_spend_logs_cleanup_cron",
|
||||
"maximum_spend_logs_retention_interval",
|
||||
"maximum_spend_logs_size",
|
||||
"maximum_spend_logs_cleanup_target_size",
|
||||
"maximum_spend_logs_size_check_interval",
|
||||
)
|
||||
_SPEND_LOG_CLEANUP_JOB_ID: Final = "spend_log_cleanup_job"
|
||||
_SPEND_LOG_SIZE_CHECK_JOB_ID: Final = "spend_log_size_check_job"
|
||||
|
||||
|
||||
def _cleanup_schedule_of(settings: Mapping[str, object]) -> tuple[object, ...]:
|
||||
return tuple(settings.get(key) for key in _CLEANUP_SCHEDULE_KEYS)
|
||||
|
||||
|
||||
def _wants_retention_job(settings: Mapping[str, object]) -> bool:
|
||||
"""The retention job runs the age rules, and the size rule too when one cron drives both."""
|
||||
return any(settings.get(key) is not None for key in _RETENTION_RULE_KEYS) or (
|
||||
settings.get("maximum_spend_logs_size") is not None and bool(settings.get("maximum_spend_logs_cleanup_cron"))
|
||||
)
|
||||
|
||||
|
||||
def _wants_size_check_job(settings: Mapping[str, object]) -> bool:
|
||||
"""The size rule gets its own job on maximum_spend_logs_size_check_interval unless a cleanup cron is set."""
|
||||
return settings.get("maximum_spend_logs_size") is not None and not settings.get("maximum_spend_logs_cleanup_cron")
|
||||
|
||||
|
||||
def _wants_cleanup_jobs(settings: Mapping[str, object]) -> bool:
|
||||
return _wants_retention_job(settings) or _wants_size_check_job(settings)
|
||||
|
||||
|
||||
@lru_cache(maxsize=4096)
|
||||
def _log_ignored_cost_map_copy(model_id: str, fields: tuple[str, ...]) -> None:
|
||||
verbose_proxy_logger.warning(
|
||||
|
|
@ -8052,39 +8077,73 @@ class ProxyConfig:
|
|||
if scheduler is None:
|
||||
return
|
||||
|
||||
wants_job: Final = any(
|
||||
general_settings.get(key) is not None
|
||||
for key in (
|
||||
"maximum_spend_logs_retention_period",
|
||||
"maximum_autorouter_session_retention_period",
|
||||
"maximum_health_check_retention_period",
|
||||
"maximum_daily_tag_spend_retention_period",
|
||||
)
|
||||
)
|
||||
if not wants_job:
|
||||
if scheduler.get_job("spend_log_cleanup_job") is not None:
|
||||
scheduler.remove_job("spend_log_cleanup_job")
|
||||
verbose_proxy_logger.info("Removed existing spend log cleanup job")
|
||||
return
|
||||
|
||||
trigger: Final = self._spend_log_cleanup_trigger()
|
||||
if trigger is None:
|
||||
return
|
||||
from litellm.proxy.db.db_transaction_queue.spend_log_cleanup import (
|
||||
SpendLogCleanup,
|
||||
)
|
||||
|
||||
scheduler.add_job(
|
||||
SpendLogCleanup().cleanup_old_spend_logs,
|
||||
trigger,
|
||||
if not _wants_retention_job(general_settings):
|
||||
if scheduler.get_job(_SPEND_LOG_CLEANUP_JOB_ID) is not None:
|
||||
scheduler.remove_job(_SPEND_LOG_CLEANUP_JOB_ID)
|
||||
verbose_proxy_logger.info("Removed existing spend log cleanup job")
|
||||
else:
|
||||
trigger: Final = self._spend_log_cleanup_trigger()
|
||||
if trigger is not None:
|
||||
scheduler.add_job(
|
||||
SpendLogCleanup().cleanup_old_spend_logs,
|
||||
trigger,
|
||||
args=[prisma_client],
|
||||
id=_SPEND_LOG_CLEANUP_JOB_ID,
|
||||
replace_existing=True,
|
||||
misfire_grace_time=APSCHEDULER_MISFIRE_GRACE_TIME,
|
||||
)
|
||||
verbose_proxy_logger.info("Spend log cleanup rescheduled with trigger: %s", trigger)
|
||||
|
||||
if not _wants_size_check_job(general_settings): # pyright: ignore[reportUnknownArgumentType] # general_settings is an untyped module dict
|
||||
if scheduler.get_job(_SPEND_LOG_SIZE_CHECK_JOB_ID) is not None: # pyright: ignore[reportUnknownMemberType] # apscheduler is untyped
|
||||
scheduler.remove_job(_SPEND_LOG_SIZE_CHECK_JOB_ID) # pyright: ignore[reportUnknownMemberType] # apscheduler is untyped
|
||||
verbose_proxy_logger.info("Removed existing spend log size check job")
|
||||
return
|
||||
size_trigger: Final = self._spend_log_size_check_trigger()
|
||||
if size_trigger is None:
|
||||
return
|
||||
scheduler.add_job( # pyright: ignore[reportUnknownMemberType] # apscheduler is untyped
|
||||
SpendLogCleanup().cleanup_spend_logs_by_size,
|
||||
size_trigger,
|
||||
args=[prisma_client],
|
||||
id="spend_log_cleanup_job",
|
||||
id=_SPEND_LOG_SIZE_CHECK_JOB_ID,
|
||||
replace_existing=True,
|
||||
misfire_grace_time=APSCHEDULER_MISFIRE_GRACE_TIME,
|
||||
)
|
||||
verbose_proxy_logger.info("Spend log cleanup rescheduled with trigger: %s", trigger)
|
||||
verbose_proxy_logger.info("Spend log size check rescheduled with trigger: %s", size_trigger)
|
||||
|
||||
def _spend_log_cleanup_trigger(self) -> BaseTrigger | None:
|
||||
def _spend_log_size_check_trigger(self) -> object | None:
|
||||
return self._staggered_interval_trigger(
|
||||
_SPEND_LOG_SIZE_CHECK_JOB_ID,
|
||||
"maximum_spend_logs_size_check_interval",
|
||||
general_settings.get("maximum_spend_logs_size_check_interval", SPEND_LOG_SIZE_CHECK_INTERVAL), # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # general_settings is an untyped module dict
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _staggered_interval_trigger(job_id: str, setting_name: str, interval: object) -> object | None:
|
||||
if not isinstance(interval, str):
|
||||
verbose_proxy_logger.error("Invalid %s value: %r", setting_name, interval)
|
||||
return None
|
||||
# this runs against a started scheduler, which the startup stagger sweep
|
||||
# cannot reach, so the offset is applied here or the job reconverges across
|
||||
# replicas the first time an admin edits the settings
|
||||
try:
|
||||
interval_seconds: Final = duration_in_seconds(interval)
|
||||
return stagger_trigger(
|
||||
job_id=job_id,
|
||||
trigger=IntervalTrigger(seconds=interval_seconds),
|
||||
period_seconds=interval_seconds,
|
||||
settings=parse_stagger_settings(general_settings), # pyright: ignore[reportUnknownArgumentType] # general_settings is an untyped module dict
|
||||
)
|
||||
except (ValueError, OverflowError):
|
||||
verbose_proxy_logger.error("Invalid %s value: %r", setting_name, interval)
|
||||
return None
|
||||
|
||||
def _spend_log_cleanup_trigger(self) -> object | None:
|
||||
cleanup_cron: Final[object] = general_settings.get("maximum_spend_logs_cleanup_cron")
|
||||
if cleanup_cron:
|
||||
from apscheduler.triggers.cron import CronTrigger
|
||||
|
|
@ -8095,24 +8154,11 @@ class ProxyConfig:
|
|||
verbose_proxy_logger.error("Invalid maximum_spend_logs_cleanup_cron value: %s", cleanup_cron)
|
||||
return None
|
||||
return cron_trigger
|
||||
retention_interval: Final[object] = general_settings.get("maximum_spend_logs_retention_interval", "1d")
|
||||
if not isinstance(retention_interval, str):
|
||||
verbose_proxy_logger.error("Invalid maximum_spend_logs_retention_interval value: %r", retention_interval)
|
||||
return None
|
||||
# this runs against a started scheduler, which the startup stagger sweep
|
||||
# cannot reach, so the offset is applied here or the job reconverges across
|
||||
# replicas the first time an admin edits the retention settings
|
||||
try:
|
||||
interval_seconds: Final = duration_in_seconds(retention_interval)
|
||||
return stagger_trigger(
|
||||
job_id="spend_log_cleanup_job",
|
||||
trigger=IntervalTrigger(seconds=interval_seconds),
|
||||
period_seconds=interval_seconds,
|
||||
settings=parse_stagger_settings(general_settings),
|
||||
)
|
||||
except (ValueError, OverflowError):
|
||||
verbose_proxy_logger.error("Invalid maximum_spend_logs_retention_interval value: %r", retention_interval)
|
||||
return None
|
||||
return self._staggered_interval_trigger(
|
||||
_SPEND_LOG_CLEANUP_JOB_ID,
|
||||
"maximum_spend_logs_retention_interval",
|
||||
general_settings.get("maximum_spend_logs_retention_interval", "1d"),
|
||||
)
|
||||
|
||||
async def _update_general_settings(self, db_general_settings: Mapping[str, SettingsJsonValue] | None) -> None:
|
||||
global general_settings
|
||||
|
|
@ -8269,8 +8315,11 @@ class ProxyConfig:
|
|||
if scheduler is not None and scheduler.state == STATE_STOPPED:
|
||||
return
|
||||
schedule: Final = self._resolved_cleanup_schedule()
|
||||
wants_job: Final = any(value is not None for value in schedule[:4])
|
||||
has_job: Final = scheduler is not None and scheduler.get_job("spend_log_cleanup_job") is not None
|
||||
wants_job: Final = _wants_cleanup_jobs(self.settings)
|
||||
has_job: Final = scheduler is not None and (
|
||||
scheduler.get_job(_SPEND_LOG_CLEANUP_JOB_ID) is not None
|
||||
or scheduler.get_job(_SPEND_LOG_SIZE_CHECK_JOB_ID) is not None # pyright: ignore[reportUnknownMemberType] # apscheduler is untyped
|
||||
)
|
||||
baseline: Final = (
|
||||
self._last_cleanup_schedule_attempt
|
||||
if has_job and self._last_cleanup_schedule_attempt is not None
|
||||
|
|
@ -11417,12 +11466,29 @@ class ProxyStartupEvent:
|
|||
|
||||
### SPEND LOG CLEANUP ###
|
||||
cleanup_settings: Final = _current_general_settings()
|
||||
if (
|
||||
cleanup_settings.get("maximum_spend_logs_retention_period") is not None
|
||||
or cleanup_settings.get("maximum_autorouter_session_retention_period") is not None
|
||||
or cleanup_settings.get("maximum_health_check_retention_period") is not None
|
||||
or cleanup_settings.get("maximum_daily_tag_spend_retention_period") is not None
|
||||
):
|
||||
if _wants_size_check_job(cleanup_settings):
|
||||
size_check_interval: Final = cleanup_settings.get(
|
||||
"maximum_spend_logs_size_check_interval", SPEND_LOG_SIZE_CHECK_INTERVAL
|
||||
)
|
||||
try:
|
||||
if not isinstance(size_check_interval, str):
|
||||
raise TypeError(size_check_interval)
|
||||
size_check_seconds: Final = duration_in_seconds(size_check_interval)
|
||||
scheduler.add_job( # pyright: ignore[reportUnknownMemberType] # apscheduler is untyped
|
||||
SpendLogCleanup().cleanup_spend_logs_by_size,
|
||||
"interval",
|
||||
seconds=size_check_seconds + random.randint(0, 60),
|
||||
args=[prisma_client],
|
||||
id=_SPEND_LOG_SIZE_CHECK_JOB_ID,
|
||||
replace_existing=True,
|
||||
misfire_grace_time=APSCHEDULER_MISFIRE_GRACE_TIME,
|
||||
)
|
||||
verbose_proxy_logger.info("Spend log size check scheduled every %s", size_check_interval)
|
||||
except (TypeError, ValueError, OverflowError):
|
||||
verbose_proxy_logger.error(
|
||||
"Invalid maximum_spend_logs_size_check_interval value: %r", size_check_interval
|
||||
)
|
||||
if _wants_retention_job(cleanup_settings):
|
||||
spend_log_cleanup: Final = SpendLogCleanup()
|
||||
cleanup_cron: Final = cleanup_settings.get("maximum_spend_logs_cleanup_cron")
|
||||
|
||||
|
|
@ -11437,7 +11503,7 @@ class ProxyStartupEvent:
|
|||
spend_log_cleanup.cleanup_old_spend_logs,
|
||||
cron_trigger,
|
||||
args=[prisma_client],
|
||||
id="spend_log_cleanup_job",
|
||||
id=_SPEND_LOG_CLEANUP_JOB_ID,
|
||||
replace_existing=True,
|
||||
misfire_grace_time=APSCHEDULER_MISFIRE_GRACE_TIME,
|
||||
)
|
||||
|
|
@ -11456,7 +11522,7 @@ class ProxyStartupEvent:
|
|||
"interval",
|
||||
seconds=interval_seconds + random.randint(0, 60),
|
||||
args=[prisma_client],
|
||||
id="spend_log_cleanup_job",
|
||||
id=_SPEND_LOG_CLEANUP_JOB_ID,
|
||||
replace_existing=True,
|
||||
misfire_grace_time=APSCHEDULER_MISFIRE_GRACE_TIME,
|
||||
)
|
||||
|
|
@ -11464,6 +11530,7 @@ class ProxyStartupEvent:
|
|||
verbose_proxy_logger.error(
|
||||
"Invalid maximum_spend_logs_retention_interval value: %r", retention_interval
|
||||
)
|
||||
if _wants_cleanup_jobs(cleanup_settings):
|
||||
proxy_config.record_cleanup_schedule_attempt(cleanup_settings)
|
||||
### CHECK BATCH COST ###
|
||||
if llm_router is not None and PROXY_BATCH_POLLING_ENABLED:
|
||||
|
|
|
|||
|
|
@ -105,3 +105,166 @@ async def test_delete_batch_survives_witnessed_lock_past_transaction_default(cap
|
|||
await database.disconnect()
|
||||
finally:
|
||||
setup.execute(sql.SQL("DROP SCHEMA {} CASCADE").format(sql.Identifier(schema)))
|
||||
|
||||
|
||||
async def test_size_rule_deletes_the_oldest_rows_until_under_the_target_and_does_not_refire_after_vacuum(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""
|
||||
Deleting rows never shrinks the table file, so the size pass must stop on live bytes, not on
|
||||
pg_total_relation_size, taking the oldest rows first and keeping the tool index on the same cutoff. Once
|
||||
autovacuum has turned the deleted rows into free space the catalog no longer shows them as dead, and the
|
||||
files still measure the same; the next check must read what the last run freed from its saved count and
|
||||
leave the table alone instead of calling the whole file live again.
|
||||
"""
|
||||
caplog.set_level(logging.INFO, logger="LiteLLM Proxy")
|
||||
schema: Final = f"integration_{uuid.uuid4().hex}"
|
||||
url: Final = os.environ["DATABASE_URL"]
|
||||
parsed: Final = urlsplit(url)
|
||||
scoped_url: Final = urlunsplit(
|
||||
parsed._replace(query=urlencode({**dict(parse_qsl(parsed.query)), "schema": schema}))
|
||||
)
|
||||
logs: Final = sql.Identifier(schema, "LiteLLM_SpendLogs")
|
||||
index: Final = sql.Identifier(schema, "LiteLLM_SpendLogToolIndex")
|
||||
config: Final = sql.Identifier(schema, "LiteLLM_Config")
|
||||
seeded: Final = 6000
|
||||
now: Final = datetime.now(timezone.utc).replace(tzinfo=None)
|
||||
stats_sql: Final = (
|
||||
"SELECT sum(n_live_tup)::int AS live, sum(n_dead_tup)::int AS dead, sum(n_tup_ins)::int AS inserted "
|
||||
"FROM pg_stat_user_tables "
|
||||
"WHERE schemaname = %s AND relname IN ('LiteLLM_SpendLogs', 'LiteLLM_SpendLogToolIndex')"
|
||||
)
|
||||
with psycopg.connect(url, autocommit=True) as setup:
|
||||
setup.execute(sql.SQL("CREATE SCHEMA {}").format(sql.Identifier(schema)))
|
||||
try:
|
||||
setup.execute(
|
||||
sql.SQL(
|
||||
'CREATE TABLE {} (request_id text, "startTime" timestamp NOT NULL, payload text, '
|
||||
'PRIMARY KEY (request_id, "startTime"))'
|
||||
).format(logs)
|
||||
)
|
||||
setup.execute(
|
||||
sql.SQL(
|
||||
"CREATE TABLE {} (request_id text, tool_name text, start_time timestamp NOT NULL, "
|
||||
"PRIMARY KEY (request_id, tool_name))"
|
||||
).format(index)
|
||||
)
|
||||
setup.execute(sql.SQL("CREATE TABLE {} (param_name text PRIMARY KEY, param_value jsonb)").format(config))
|
||||
setup.execute(
|
||||
sql.SQL(
|
||||
"INSERT INTO {} SELECT 'req-' || lpad(i::text, 6, '0'), %s - (%s - i) * interval '1 minute', "
|
||||
"repeat('x', 1000) FROM generate_series(1, %s) AS i"
|
||||
).format(logs),
|
||||
(now, seeded, seeded),
|
||||
)
|
||||
setup.execute(
|
||||
sql.SQL(
|
||||
"INSERT INTO {} SELECT request_id, 'tool', \"startTime\" FROM {} WHERE right(request_id, 1) = '0'"
|
||||
).format(index, logs)
|
||||
)
|
||||
index_rows: Final = seeded // 10
|
||||
inserts_deadline: Final = time.monotonic() + 10
|
||||
while True:
|
||||
# a backend only ships its pending row counters when it runs another command, so give the
|
||||
# seeding connection one; the size rule rolls its count forward from n_tup_ins, which must
|
||||
# show the seed inserts before ANALYZE sets the live estimate or the two add up to double
|
||||
setup.execute("SELECT 1")
|
||||
(inserts,) = read_rows(stats_sql, (schema,))
|
||||
if inserts["inserted"] == seeded + index_rows:
|
||||
break
|
||||
assert time.monotonic() < inserts_deadline, f"row statistics never saw the seed inserts: {inserts}"
|
||||
await asyncio.sleep(0.1)
|
||||
setup.execute(sql.SQL("ANALYZE {}").format(logs))
|
||||
setup.execute(sql.SQL("ANALYZE {}").format(index))
|
||||
stats_deadline: Final = time.monotonic() + 10
|
||||
while True:
|
||||
setup.execute("SELECT 1")
|
||||
(stats,) = read_rows(stats_sql, (schema,))
|
||||
if stats["live"] == seeded + index_rows:
|
||||
break
|
||||
assert time.monotonic() < stats_deadline, f"row statistics never caught up with ANALYZE: {stats}"
|
||||
await asyncio.sleep(0.1)
|
||||
(allocated,) = read_rows(
|
||||
"SELECT (pg_total_relation_size(%s::regclass) + pg_total_relation_size(%s::regclass))::bigint "
|
||||
"AS allocated",
|
||||
(logs.as_string(setup), index.as_string(setup)),
|
||||
)
|
||||
allocated_before: Final = int(str(allocated["allocated"]))
|
||||
assert allocated_before > 0
|
||||
|
||||
cleanup: Final = SpendLogCleanup(
|
||||
general_settings={
|
||||
# trigger below what is there, clean down to half of it
|
||||
"maximum_spend_logs_size": str(allocated_before * 3 // 4),
|
||||
"maximum_spend_logs_cleanup_target_size": str(allocated_before // 2),
|
||||
"maximum_spend_logs_cleanup_batch_size": 100,
|
||||
}
|
||||
)
|
||||
cleanup.pod_lock_manager = None
|
||||
|
||||
async def size_check() -> None:
|
||||
database: Final = Prisma(datasource={"url": scoped_url})
|
||||
await database.connect()
|
||||
try:
|
||||
await cleanup.cleanup_spend_logs_by_size(PartitionConnection(database))
|
||||
finally:
|
||||
await database.disconnect()
|
||||
|
||||
await size_check()
|
||||
|
||||
remaining: Final = read_rows(
|
||||
f'SELECT request_id FROM {logs.as_string(setup)} ORDER BY "startTime"',
|
||||
(),
|
||||
)
|
||||
# half the bytes is half the rows: the oldest 3000 spend logs and their 300 index rows go, counted
|
||||
# from the deletes themselves, so the stats collector's lag cannot make the run overshoot. A page the
|
||||
# files grew by between this measurement and the run's probe costs at most one more batch
|
||||
assert 2900 <= len(remaining) <= 3000, f"{len(remaining)} spend logs remain, expected 3000 less one batch"
|
||||
assert remaining[0]["request_id"] == f"req-{seeded - len(remaining) + 1:06d}"
|
||||
assert remaining[-1]["request_id"] == f"req-{seeded:06d}"
|
||||
(orphans,) = read_rows(
|
||||
f"SELECT count(*)::int AS n FROM {index.as_string(setup)} i "
|
||||
f"WHERE NOT EXISTS (SELECT 1 FROM {logs.as_string(setup)} l WHERE l.request_id = i.request_id)",
|
||||
(),
|
||||
)
|
||||
assert orphans["n"] == 0
|
||||
(kept_index,) = read_rows(f"SELECT count(*)::int AS n FROM {index.as_string(setup)}", ())
|
||||
assert kept_index["n"] == len(remaining) // 10
|
||||
summary: Final = next(
|
||||
record.getMessage()
|
||||
for record in caplog.records
|
||||
if "Spend log cleanup run finished" in record.getMessage()
|
||||
)
|
||||
assert "target_reached=true" in summary
|
||||
assert f"allocated_before={allocated_before}" in summary
|
||||
# the file did not shrink: that is the whole reason the reclaimable count exists
|
||||
assert f"allocated_after={allocated_before}" in summary
|
||||
assert "VACUUM FULL or pg_repack" in summary
|
||||
|
||||
deletes_deadline: Final = time.monotonic() + 10
|
||||
while True:
|
||||
(after_deletes,) = read_rows(stats_sql, (schema,))
|
||||
if after_deletes["dead"] == 3300:
|
||||
break
|
||||
assert time.monotonic() < deletes_deadline, f"row statistics never saw the deletes: {after_deletes}"
|
||||
await asyncio.sleep(0.1)
|
||||
setup.execute(sql.SQL("VACUUM (ANALYZE) {}").format(logs))
|
||||
setup.execute(sql.SQL("VACUUM (ANALYZE) {}").format(index))
|
||||
vacuum_deadline: Final = time.monotonic() + 10
|
||||
while True:
|
||||
(after_vacuum,) = read_rows(stats_sql, (schema,))
|
||||
if after_vacuum["dead"] == 0:
|
||||
break
|
||||
assert time.monotonic() < vacuum_deadline, f"row statistics never caught up with VACUUM: {after_vacuum}"
|
||||
await asyncio.sleep(0.1)
|
||||
caplog.clear()
|
||||
|
||||
await size_check()
|
||||
|
||||
(still_there,) = read_rows(f"SELECT count(*)::int AS n FROM {logs.as_string(setup)}", ())
|
||||
assert still_there["n"] == len(remaining), (
|
||||
"the second check deleted rows the first one had already brought under the limit"
|
||||
)
|
||||
assert any("is under maximum_spend_logs_size" in record.getMessage() for record in caplog.records)
|
||||
finally:
|
||||
setup.execute(sql.SQL("DROP SCHEMA {} CASCADE").format(sql.Identifier(schema)))
|
||||
|
|
|
|||
|
|
@ -1681,7 +1681,7 @@ async def test_a_cleanup_delete_batch_renders_a_postgres_delete_span_for_its_tab
|
|||
assert await postgres_span_names() == ("postgres.delete LiteLLM_SpendLogs",)
|
||||
|
||||
|
||||
# --- spend log storage measurement and the reclaimable-bytes state ---------------------------------------
|
||||
# --- size rule (maximum_spend_logs_size) ---------------------------------------------------------------
|
||||
|
||||
|
||||
def _storage(allocated: int, live: int, dead: int, inserted: int | None = None, deleted: int | None = None) -> dict:
|
||||
|
|
@ -1694,6 +1694,50 @@ def _storage(allocated: int, live: int, dead: int, inserted: int | None = None,
|
|||
}
|
||||
|
||||
|
||||
def _wire_size_probes(
|
||||
db, storage_rows: list[dict], cutoff: datetime, saved_state: SizeCleanupState | None = None
|
||||
) -> list[SizeCleanupState]:
|
||||
"""
|
||||
Answer the size pass's catalog reads: the storage probe walks `storage_rows` one measurement per call
|
||||
(repeating the last), the step-cutoff query returns `cutoff`, the remaining-rows probe returns 0, and the
|
||||
LiteLLM_Config row holds `saved_state`. Writes of that row land in the returned list instead of
|
||||
db.execute_raw, so DELETE counts stay what tests configure.
|
||||
"""
|
||||
measurements = iter(storage_rows)
|
||||
last = [storage_rows[-1]]
|
||||
saves: list[SizeCleanupState] = []
|
||||
deletes = db.execute_raw
|
||||
|
||||
async def _query_raw(sql, *args):
|
||||
if "pg_total_relation_size" in sql:
|
||||
row = next(measurements, None)
|
||||
if row is None:
|
||||
row = last[0]
|
||||
return [row]
|
||||
if "OFFSET" in sql:
|
||||
return [{"startTime": cutoff}]
|
||||
if '"LiteLLM_Config"' in sql:
|
||||
return [{"state_json": saved_state.model_dump_json()}] if saved_state is not None else []
|
||||
return [{"remaining": 0}]
|
||||
|
||||
async def _execute_raw(sql, *args):
|
||||
if '"LiteLLM_Config"' in sql:
|
||||
saves.append(SizeCleanupState.model_validate_json(args[1]))
|
||||
return 1
|
||||
return await deletes(sql, *args)
|
||||
|
||||
db.query_raw = AsyncMock(side_effect=_query_raw)
|
||||
db.execute_raw = _execute_raw
|
||||
db.deletes = deletes
|
||||
return saves
|
||||
|
||||
|
||||
def _size_cleaner(settings: dict, **kwargs) -> SpendLogCleanup:
|
||||
cleaner = SpendLogCleanup(general_settings={"maximum_spend_logs_size": "1GB", **settings}, **kwargs)
|
||||
cleaner.pod_lock_manager = None
|
||||
return cleaner
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("value", "expected"),
|
||||
[
|
||||
|
|
@ -1719,6 +1763,160 @@ def test_size_in_bytes_rejects_what_cannot_be_a_size(value):
|
|||
size_in_bytes(value)
|
||||
|
||||
|
||||
def test_an_unset_size_means_no_size_rule_and_a_bad_target_falls_back_to_the_size(caplog):
|
||||
assert SpendLogCleanup(general_settings={})._size_rule() is None
|
||||
rule = _size_cleaner({"maximum_spend_logs_cleanup_target_size": "200MB"})._size_rule()
|
||||
assert rule is not None
|
||||
assert (rule.limit_bytes, rule.target_bytes) == (1024**3, 200 * 1024**2)
|
||||
with caplog.at_level(logging.ERROR, logger="LiteLLM Proxy"):
|
||||
rule_without_room = _size_cleaner({"maximum_spend_logs_cleanup_target_size": "2GB"})._size_rule()
|
||||
assert rule_without_room is not None
|
||||
assert rule_without_room.target_bytes == rule_without_room.limit_bytes == 1024**3
|
||||
assert any("must be below maximum_spend_logs_size" in record.getMessage() for record in caplog.records)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_size_pass_under_the_limit_issues_no_delete():
|
||||
db = MagicMock()
|
||||
_wire_tx(db)
|
||||
db.execute_raw = AsyncMock(return_value=0)
|
||||
_wire_size_probes(db, [_storage(allocated=500 * 1024**2, live=1000, dead=0)], cutoff=datetime.now(timezone.utc))
|
||||
client = MagicMock()
|
||||
client.db = db
|
||||
|
||||
await _size_cleaner({}).cleanup_spend_logs_by_size(client)
|
||||
|
||||
assert db.deletes.await_count == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_size_pass_deletes_oldest_first_in_steps_until_the_live_estimate_is_under_the_target(caplog):
|
||||
"""
|
||||
Over the limit, the pass turns the next step of oldest rows into a cutoff and runs the retention DELETE
|
||||
for the spend logs and their tool-index rows, re-measuring after each step; it stops on the first
|
||||
measurement under the target, not when the file size (which deletes never shrink) moves.
|
||||
"""
|
||||
db = MagicMock()
|
||||
_wire_tx(db)
|
||||
# 3 GiB over 3000 rows is 1 MiB a row, so 2560 rows take it from 3 GiB to the 512 MiB target: one step
|
||||
# of 3 batches; the tool index is cleaned on the same cutoff
|
||||
db.execute_raw = AsyncMock(side_effect=[1000, 1000, 1000, 300, 0])
|
||||
step_cutoff = datetime(2026, 9, 1, tzinfo=timezone.utc)
|
||||
allocated = 3 * 1024**3
|
||||
_wire_size_probes(db, [_storage(allocated, live=3000, dead=0)], cutoff=step_cutoff)
|
||||
client = MagicMock()
|
||||
client.db = db
|
||||
cleaner = _size_cleaner({"maximum_spend_logs_cleanup_target_size": "512MB"})
|
||||
|
||||
with caplog.at_level(logging.INFO, logger="LiteLLM Proxy"):
|
||||
await cleaner.cleanup_spend_logs_by_size(client)
|
||||
|
||||
deletes = [(call[0][0], call[0][1:]) for call in db.deletes.call_args_list]
|
||||
assert len(deletes) == 5
|
||||
assert 'DELETE FROM "LiteLLM_SpendLogs"' in deletes[0][0]
|
||||
assert 'WHERE "startTime" < $1' in deletes[0][0]
|
||||
assert deletes[0][1][0] == step_cutoff
|
||||
assert 'DELETE FROM "LiteLLM_SpendLogToolIndex"' in deletes[3][0]
|
||||
assert deletes[3][1][0] == step_cutoff
|
||||
offset_query = next(call for call in db.query_raw.call_args_list if "OFFSET" in call[0][0])
|
||||
assert offset_query[0][1] == 3000
|
||||
summary = next(r.getMessage() for r in caplog.records if "Spend log cleanup run finished" in r.getMessage())
|
||||
assert "outcome=completed" in summary
|
||||
assert "LiteLLM_SpendLogs (size rule): deleted=3000 stop_reason=target_reached" in summary
|
||||
assert "LiteLLM_SpendLogToolIndex (size rule): deleted=300 stop_reason=target_reached" in summary
|
||||
assert f"allocated_before={allocated}" in summary
|
||||
assert "estimated_live_before=3221225472" in summary
|
||||
assert "estimated_live_after=0" in summary
|
||||
assert "target_reached=true above_target=0" in summary
|
||||
assert "VACUUM FULL or pg_repack" in summary
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_size_pass_stops_at_the_run_batch_cap_and_reports_how_far_above_the_target_it_is(caplog):
|
||||
db = MagicMock()
|
||||
_wire_tx(db)
|
||||
# 4 GiB over 4000 rows: 3584 rows must go to reach the 512 MiB target, the cap allows 3 batches
|
||||
db.execute_raw = AsyncMock(side_effect=[1000, 1000, 1000, 0])
|
||||
allocated = 4 * 1024**3
|
||||
saves = _wire_size_probes(db, [_storage(allocated, live=4000, dead=0)], cutoff=datetime.now(timezone.utc))
|
||||
client = MagicMock()
|
||||
client.db = db
|
||||
cleaner = _size_cleaner(
|
||||
{"maximum_spend_logs_cleanup_target_size": "512MB", "maximum_spend_logs_cleanup_max_batches": 3}
|
||||
)
|
||||
|
||||
with caplog.at_level(logging.WARNING, logger="LiteLLM Proxy"):
|
||||
await cleaner.cleanup_spend_logs_by_size(client)
|
||||
|
||||
assert db.deletes.await_count == 4
|
||||
summary = next(r for r in caplog.records if "Spend log cleanup run finished" in r.getMessage())
|
||||
assert summary.levelno == logging.WARNING
|
||||
assert "outcome=batch_cap_reached" in summary.getMessage()
|
||||
assert "LiteLLM_SpendLogs (size rule): deleted=3000 stop_reason=batch_cap_reached" in summary.getMessage()
|
||||
# 1000 of 4000 rows left is 1 GiB live, 512 MiB above the target
|
||||
assert f"target_reached=false above_target={512 * 1024**2}" in summary.getMessage()
|
||||
# the next check starts from what this run freed, with the deletes already counted out of n_tup_del
|
||||
assert saves == [
|
||||
SizeCleanupState(
|
||||
allocated_bytes=allocated,
|
||||
reclaimable_bytes=round(3000 * allocated / 4000),
|
||||
bytes_per_row=allocated / 4000,
|
||||
inserted_total=4000,
|
||||
deleted_total=3000,
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_size_pass_skips_on_stale_row_statistics_instead_of_trusting_an_estimate_of_zero(caplog):
|
||||
db = MagicMock()
|
||||
_wire_tx(db)
|
||||
db.execute_raw = AsyncMock(return_value=1000)
|
||||
_wire_size_probes(db, [_storage(allocated=5 * 1024**3, live=0, dead=0)], cutoff=datetime.now(timezone.utc))
|
||||
client = MagicMock()
|
||||
client.db = db
|
||||
|
||||
with caplog.at_level(logging.WARNING, logger="LiteLLM Proxy"):
|
||||
await _size_cleaner({}).cleanup_spend_logs_by_size(client)
|
||||
|
||||
assert db.deletes.await_count == 0
|
||||
assert any("row statistics are stale" in record.getMessage() for record in caplog.records)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_size_pass_does_not_refire_after_autovacuum_forgets_the_rows_it_deleted():
|
||||
"""
|
||||
The run before this one deleted 15,000 of 20,000 rows and saved what it freed. Autovacuum has since
|
||||
turned those dead tuples into free space, so pg_stat_user_tables reports 5,000 live and 0 dead rows while
|
||||
the files still measure the full 32 MB. Trusting the live share would call all 32 MB live and empty the
|
||||
table; the saved count says 8 MB are live, under the limit, so nothing is deleted.
|
||||
"""
|
||||
db = MagicMock()
|
||||
_wire_tx(db)
|
||||
db.execute_raw = AsyncMock(return_value=1000)
|
||||
after_last_run = SizeCleanupState(
|
||||
allocated_bytes=32_030_720,
|
||||
reclaimable_bytes=24_015_000,
|
||||
bytes_per_row=1601.0,
|
||||
inserted_total=20_000,
|
||||
deleted_total=15_000,
|
||||
)
|
||||
saves = _wire_size_probes(
|
||||
db,
|
||||
[_storage(32_030_720, live=5000, dead=0, inserted=20_000, deleted=15_000)],
|
||||
cutoff=datetime.now(timezone.utc),
|
||||
saved_state=after_last_run,
|
||||
)
|
||||
client = MagicMock()
|
||||
client.db = db
|
||||
cleaner = _size_cleaner({"maximum_spend_logs_size": "16MB", "maximum_spend_logs_cleanup_target_size": "10MB"})
|
||||
|
||||
await cleaner.cleanup_spend_logs_by_size(client)
|
||||
|
||||
assert db.deletes.await_count == 0
|
||||
assert saves[0].reclaimable_bytes == 24_015_000
|
||||
|
||||
|
||||
_RUN_STATE = SizeCleanupState(
|
||||
allocated_bytes=32_030_720,
|
||||
reclaimable_bytes=24_015_000,
|
||||
|
|
@ -1765,6 +1963,58 @@ def test_advance_size_state_tracks_free_space_through_vacuum_inserts_and_other_d
|
|||
assert 0 <= probe["allocated_bytes"] - advanced.reclaimable_bytes <= probe["allocated_bytes"], name
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_size_only_run_leaves_the_age_rules_to_the_retention_job():
|
||||
db = MagicMock()
|
||||
_wire_tx(db)
|
||||
db.execute_raw = AsyncMock(return_value=0)
|
||||
_wire_size_probes(db, [_storage(allocated=10 * 1024**3, live=100, dead=0)], cutoff=datetime.now(timezone.utc))
|
||||
client = MagicMock()
|
||||
client.db = db
|
||||
cleaner = _size_cleaner(
|
||||
{"maximum_spend_logs_retention_period": "7d", "maximum_health_check_retention_period": "30d"}
|
||||
)
|
||||
|
||||
await cleaner.cleanup_spend_logs_by_size(client)
|
||||
|
||||
tables = {call[0][0].split('"')[1] for call in db.deletes.call_args_list}
|
||||
assert tables == {"LiteLLM_SpendLogs", "LiteLLM_SpendLogToolIndex"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(("cron", "size_pass_expected"), [(None, False), ("0 3 * * *", True)])
|
||||
async def test_the_retention_run_carries_the_size_pass_only_when_one_cron_drives_both(cron, size_pass_expected):
|
||||
db = MagicMock()
|
||||
_wire_tx(db)
|
||||
db.execute_raw = AsyncMock(return_value=0)
|
||||
_wire_size_probes(db, [_storage(allocated=10 * 1024**3, live=100, dead=0)], cutoff=datetime.now(timezone.utc))
|
||||
client = MagicMock()
|
||||
client.db = db
|
||||
cleaner = _size_cleaner({"maximum_spend_logs_retention_period": "7d", "maximum_spend_logs_cleanup_cron": cron})
|
||||
|
||||
await cleaner.cleanup_old_spend_logs(client)
|
||||
|
||||
measured = any("pg_total_relation_size" in call[0][0] for call in db.query_raw.call_args_list)
|
||||
assert measured is size_pass_expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("settings", "retention_job", "size_job"),
|
||||
[
|
||||
({}, False, False),
|
||||
({"maximum_spend_logs_retention_period": "7d"}, True, False),
|
||||
({"maximum_spend_logs_size": "180GB"}, False, True),
|
||||
({"maximum_spend_logs_size": "180GB", "maximum_spend_logs_retention_period": "7d"}, True, True),
|
||||
({"maximum_spend_logs_size": "180GB", "maximum_spend_logs_cleanup_cron": "0 3 * * *"}, True, False),
|
||||
],
|
||||
)
|
||||
def test_which_scheduler_jobs_the_cleanup_settings_ask_for(settings, retention_job, size_job):
|
||||
from litellm.proxy.proxy_server import _wants_retention_job, _wants_size_check_job
|
||||
|
||||
assert _wants_retention_job(settings) is retention_job
|
||||
assert _wants_size_check_job(settings) is size_job
|
||||
|
||||
|
||||
def test_after_deleting_counts_the_rows_and_any_file_growth_as_reclaimable_and_to_state_keeps_the_counters():
|
||||
storage = SpendLogStorage(allocated_bytes=32_030_720, reclaimable_bytes=0, bytes_per_row=1601.0, live_rows=20_000)
|
||||
|
||||
|
|
|
|||
15
ui/litellm-dashboard/src/lib/http/schema.d.ts
generated
vendored
15
ui/litellm-dashboard/src/lib/http/schema.d.ts
generated
vendored
|
|
@ -30072,11 +30072,26 @@ export interface components {
|
|||
* @description Wall-clock budget for one spend log cleanup run (e.g. '5m'), shared across every table it prunes. A run that hits the budget stops and the next run resumes from where it left off. Defaults to '5m'.
|
||||
*/
|
||||
maximum_spend_logs_cleanup_run_budget?: string | null;
|
||||
/**
|
||||
* Maximum Spend Logs Cleanup Target Size
|
||||
* @description Level a size-triggered cleanup run deletes down to (e.g. '20GB'). Must be below maximum_spend_logs_size. Unset means the run stops as soon as the tables are back under maximum_spend_logs_size.
|
||||
*/
|
||||
maximum_spend_logs_cleanup_target_size?: string | null;
|
||||
/**
|
||||
* Maximum Spend Logs Retention Period
|
||||
* @description Maximum retention period for spend logs (e.g., '7d' for 7 days). Logs older than this will be deleted.
|
||||
*/
|
||||
maximum_spend_logs_retention_period?: string | null;
|
||||
/**
|
||||
* Maximum Spend Logs Size
|
||||
* @description Size rule for spend logs (e.g. '180GB'): when LiteLLM_SpendLogs plus LiteLLM_SpendLogToolIndex (heap, indexes and TOAST, measured with pg_total_relation_size) hold more live data than this, the cleanup job deletes the oldest spend logs first until they are under maximum_spend_logs_cleanup_target_size. Independent of maximum_spend_logs_retention_period: a row is deleted when either rule says so. Unset means no size rule.
|
||||
*/
|
||||
maximum_spend_logs_size?: string | null;
|
||||
/**
|
||||
* Maximum Spend Logs Size Check Interval
|
||||
* @description How often the size rule is checked (e.g. '10m'). Defaults to '1d'. Ignored when maximum_spend_logs_cleanup_cron is set, which then drives both the age and the size cleanup.
|
||||
*/
|
||||
maximum_spend_logs_size_check_interval?: string | null;
|
||||
/**
|
||||
* Mcp Advertised Versions
|
||||
* @description MCP revisions enabled by the gateway. Defaults to all completed legacy revisions. Modern protocol serving requires explicit opt-in. Apps/Tasks remain disabled.
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue