mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-21 00:21:49 +00:00
feat(proxy): add LiteLLM_DailyGlobalSpend key-free rollup for the usage dashboard
Adds a daily spend table without api_key or user_id, written atomically alongside LiteLLM_DailyUserSpend from the batched writer, reconciled from history by a scheduled job that advances a marker in LiteLLM_Config, and read by the key-free arm of the aggregated usage query once the marker covers the requested range. Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yassin
parent
c3e937b845
commit
c8a2d8c349
15 changed files with 1347 additions and 32 deletions
|
|
@ -0,0 +1,33 @@
|
|||
-- CreateTable
|
||||
CREATE TABLE IF NOT EXISTS "LiteLLM_DailyGlobalSpend" (
|
||||
"id" TEXT NOT NULL,
|
||||
"date" TEXT NOT NULL,
|
||||
"model" TEXT,
|
||||
"model_group" TEXT,
|
||||
"custom_llm_provider" TEXT,
|
||||
"mcp_namespaced_tool_name" TEXT,
|
||||
"endpoint" TEXT,
|
||||
"prompt_tokens" BIGINT NOT NULL DEFAULT 0,
|
||||
"completion_tokens" BIGINT NOT NULL DEFAULT 0,
|
||||
"cache_read_input_tokens" BIGINT NOT NULL DEFAULT 0,
|
||||
"cache_creation_input_tokens" BIGINT NOT NULL DEFAULT 0,
|
||||
"compression_saved_tokens" BIGINT NOT NULL DEFAULT 0,
|
||||
"compression_savings_spend" DOUBLE PRECISION NOT NULL DEFAULT 0.0,
|
||||
"prompt_caching_savings_spend" DOUBLE PRECISION NOT NULL DEFAULT 0.0,
|
||||
"gateway_injected_caching_savings_spend" DOUBLE PRECISION NOT NULL DEFAULT 0.0,
|
||||
"autorouter_savings_spend" DOUBLE PRECISION NOT NULL DEFAULT 0.0,
|
||||
"spend" DOUBLE PRECISION NOT NULL DEFAULT 0.0,
|
||||
"api_requests" BIGINT NOT NULL DEFAULT 0,
|
||||
"successful_requests" BIGINT NOT NULL DEFAULT 0,
|
||||
"failed_requests" BIGINT NOT NULL DEFAULT 0,
|
||||
"created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
"updated_at" TIMESTAMP(3) NOT NULL,
|
||||
|
||||
CONSTRAINT "LiteLLM_DailyGlobalSpend_pkey" PRIMARY KEY ("id")
|
||||
);
|
||||
|
||||
-- CreateIndex
|
||||
CREATE INDEX IF NOT EXISTS "LiteLLM_DailyGlobalSpend_date_idx" ON "LiteLLM_DailyGlobalSpend"("date");
|
||||
|
||||
-- CreateIndex
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS "LiteLLM_DailyGlobalSpend_date_model_model_group_custom_llm__key" ON "LiteLLM_DailyGlobalSpend"("date", "model", "model_group", "custom_llm_provider", "mcp_namespaced_tool_name", "endpoint");
|
||||
|
|
@ -809,6 +809,35 @@ model LiteLLM_DailyUserSpend {
|
|||
@@index([endpoint])
|
||||
}
|
||||
|
||||
// Key-free daily rollup of LiteLLM_DailyUserSpend, read by the global usage view
|
||||
model LiteLLM_DailyGlobalSpend {
|
||||
id String @id @default(uuid())
|
||||
date String
|
||||
model String?
|
||||
model_group String?
|
||||
custom_llm_provider String?
|
||||
mcp_namespaced_tool_name String?
|
||||
endpoint String?
|
||||
prompt_tokens BigInt @default(0)
|
||||
completion_tokens BigInt @default(0)
|
||||
cache_read_input_tokens BigInt @default(0)
|
||||
cache_creation_input_tokens BigInt @default(0)
|
||||
compression_saved_tokens BigInt @default(0)
|
||||
compression_savings_spend Float @default(0.0)
|
||||
prompt_caching_savings_spend Float @default(0.0)
|
||||
gateway_injected_caching_savings_spend Float @default(0.0)
|
||||
autorouter_savings_spend Float @default(0.0)
|
||||
spend Float @default(0.0)
|
||||
api_requests BigInt @default(0)
|
||||
successful_requests BigInt @default(0)
|
||||
failed_requests BigInt @default(0)
|
||||
created_at DateTime @default(now())
|
||||
updated_at DateTime @updatedAt
|
||||
|
||||
@@unique([date, model, model_group, custom_llm_provider, mcp_namespaced_tool_name, endpoint])
|
||||
@@index([date])
|
||||
}
|
||||
|
||||
// Track daily organization spend metrics per model and key
|
||||
model LiteLLM_DailyOrganizationSpend {
|
||||
id String @id @default(uuid())
|
||||
|
|
|
|||
|
|
@ -2034,6 +2034,9 @@ PTU_ROLLUP_MAX_BACKFILL_DAYS: Final[int] = 90
|
|||
# Deployments named in the lapsed-window alert before it is truncated, so a fleet-wide
|
||||
# expiry cannot produce an alert too large for the channel delivering it.
|
||||
PTU_LAPSED_ALERT_LIMIT: Final[int] = 10
|
||||
DAILY_GLOBAL_SPEND_RECONCILE_JOB_ID: Final[str] = "daily_global_spend_reconcile_job"
|
||||
DAILY_GLOBAL_SPEND_RECONCILE_LOCK_TTL_SECONDS: Final[int] = 3600
|
||||
DAILY_GLOBAL_SPEND_RECONCILED_THROUGH_PARAM: Final[str] = "daily_global_spend_reconciled_through"
|
||||
# Slack allowed when deciding a sentinel row is stale. The row's updated_at and the
|
||||
# run's cutoff are stamped by different hosts, so clock skew between them must not let
|
||||
# one run delete a charge another just wrote. A stale row is hours old and a concurrent
|
||||
|
|
|
|||
|
|
@ -25,29 +25,41 @@ SpendRow = Mapping[str, object]
|
|||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class DailySpendTable:
|
||||
"""The physical table behind one entity's daily rollup."""
|
||||
"""A daily rollup table and the unique constraint its upserts arbitrate on."""
|
||||
|
||||
name: str
|
||||
entity_id_column: str
|
||||
key_columns: tuple[str, ...]
|
||||
carries_request_id: bool = False
|
||||
|
||||
|
||||
DAILY_SPEND_TABLES: Final[Mapping[DailySpendEntity, DailySpendTable]] = MappingProxyType(
|
||||
{
|
||||
"user": DailySpendTable(name="LiteLLM_DailyUserSpend", entity_id_column="user_id"),
|
||||
"team": DailySpendTable(name="LiteLLM_DailyTeamSpend", entity_id_column="team_id"),
|
||||
"org": DailySpendTable(name="LiteLLM_DailyOrganizationSpend", entity_id_column="organization_id"),
|
||||
"end_user": DailySpendTable(name="LiteLLM_DailyEndUserSpend", entity_id_column="end_user_id"),
|
||||
"agent": DailySpendTable(name="LiteLLM_DailyAgentSpend", entity_id_column="agent_id"),
|
||||
"tag": DailySpendTable(name="LiteLLM_DailyTagSpend", entity_id_column="tag", carries_request_id=True),
|
||||
}
|
||||
)
|
||||
|
||||
# The unique constraint's columns after the entity id, in constraint order. A NULL can
|
||||
# never match itself in a unique index, so every one of these is normalized to '': the
|
||||
# conflict target has to be NULL-free or the row is re-inserted on every single flush.
|
||||
_KEY_COLUMNS: Final = ("date", "api_key", "model", "custom_llm_provider", "mcp_namespaced_tool_name", "endpoint")
|
||||
|
||||
|
||||
def _entity_table(name: str, entity_id_column: str, carries_request_id: bool = False) -> DailySpendTable:
|
||||
return DailySpendTable(
|
||||
name=name, key_columns=(entity_id_column, *_KEY_COLUMNS), carries_request_id=carries_request_id
|
||||
)
|
||||
|
||||
|
||||
DAILY_SPEND_TABLES: Final[Mapping[DailySpendEntity, DailySpendTable]] = MappingProxyType(
|
||||
{
|
||||
"user": _entity_table("LiteLLM_DailyUserSpend", "user_id"),
|
||||
"team": _entity_table("LiteLLM_DailyTeamSpend", "team_id"),
|
||||
"org": _entity_table("LiteLLM_DailyOrganizationSpend", "organization_id"),
|
||||
"end_user": _entity_table("LiteLLM_DailyEndUserSpend", "end_user_id"),
|
||||
"agent": _entity_table("LiteLLM_DailyAgentSpend", "agent_id"),
|
||||
"tag": _entity_table("LiteLLM_DailyTagSpend", "tag", carries_request_id=True),
|
||||
}
|
||||
)
|
||||
|
||||
GLOBAL_SPEND_TABLE: Final = DailySpendTable(
|
||||
name="LiteLLM_DailyGlobalSpend",
|
||||
key_columns=("date", "model", "model_group", "custom_llm_provider", "mcp_namespaced_tool_name", "endpoint"),
|
||||
)
|
||||
|
||||
_COUNTER_COLUMNS: Final = (
|
||||
"prompt_tokens",
|
||||
"completion_tokens",
|
||||
|
|
@ -92,7 +104,7 @@ def _as_float(value: object) -> float:
|
|||
|
||||
def conflict_key(table: DailySpendTable, transaction: SpendRow) -> tuple[str, ...]:
|
||||
"""The tuple the database arbitrates the upsert on, normalized free of NULLs."""
|
||||
return tuple(_as_text(transaction.get(column)) for column in (table.entity_id_column, *_KEY_COLUMNS))
|
||||
return tuple(_as_text(transaction.get(column)) for column in table.key_columns)
|
||||
|
||||
|
||||
def _merge(group: Sequence[SpendRow]) -> SpendRow:
|
||||
|
|
@ -130,7 +142,11 @@ def _row_params(
|
|||
return (
|
||||
str(uuid.uuid4()),
|
||||
*key,
|
||||
None if transaction.get("model_group") is None else _as_text(transaction.get("model_group")),
|
||||
*(
|
||||
()
|
||||
if "model_group" in table.key_columns
|
||||
else (None if transaction.get("model_group") is None else _as_text(transaction.get("model_group")),)
|
||||
),
|
||||
*(_as_int(transaction.get(column)) for column in _COUNTER_COLUMNS),
|
||||
*(_as_float(transaction.get(column)) for column in _SPEND_COLUMNS),
|
||||
*((None if request_id is None else _as_text(request_id),) if table.carries_request_id else ()),
|
||||
|
|
@ -140,26 +156,25 @@ def _row_params(
|
|||
def _insert_columns(table: DailySpendTable) -> tuple[str, ...]:
|
||||
return (
|
||||
"id",
|
||||
table.entity_id_column,
|
||||
*_KEY_COLUMNS,
|
||||
"model_group",
|
||||
*table.key_columns,
|
||||
*(() if "model_group" in table.key_columns else ("model_group",)),
|
||||
*_COUNTER_COLUMNS,
|
||||
*_SPEND_COLUMNS,
|
||||
*(("request_id",) if table.carries_request_id else ()),
|
||||
)
|
||||
|
||||
|
||||
def build_bulk_upsert(
|
||||
def _upsert_statement(
|
||||
table: DailySpendTable,
|
||||
batch: Sequence[tuple[tuple[str, ...], SpendRow]],
|
||||
) -> tuple[str, tuple[SqlValue, ...]]:
|
||||
"""The single statement writing one merged batch, plus its positional arguments."""
|
||||
first_param: int,
|
||||
) -> str:
|
||||
columns: Final = _insert_columns(table)
|
||||
quoted_table: Final = f'"{table.name}"'
|
||||
rows: Final = ", ".join(
|
||||
"("
|
||||
+ ", ".join(
|
||||
f"${row_index * len(columns) + offset + 1}::{_CASTS.get(column, 'text')}"
|
||||
f"${first_param + row_index * len(columns) + offset}::{_CASTS.get(column, 'text')}"
|
||||
for offset, column in enumerate(columns)
|
||||
)
|
||||
+ ", (NOW() AT TIME ZONE 'UTC'))"
|
||||
|
|
@ -176,11 +191,44 @@ def build_bulk_upsert(
|
|||
if table.carries_request_id
|
||||
else ""
|
||||
)
|
||||
sql: Final = (
|
||||
return (
|
||||
f'INSERT INTO {quoted_table} ({_quoted(columns)}, "updated_at")\n'
|
||||
f"VALUES {rows}\n"
|
||||
f"ON CONFLICT ({_quoted((table.entity_id_column, *_KEY_COLUMNS))}) DO UPDATE SET\n"
|
||||
f"ON CONFLICT ({_quoted(table.key_columns)}) DO UPDATE SET\n"
|
||||
f" {increments}{request_id_update},\n"
|
||||
f" \"updated_at\" = (NOW() AT TIME ZONE 'UTC')"
|
||||
)
|
||||
return sql, tuple(value for key, transaction in batch for value in _row_params(table, key, transaction))
|
||||
|
||||
|
||||
def _params(table: DailySpendTable, batch: Sequence[tuple[tuple[str, ...], SpendRow]]) -> tuple[SqlValue, ...]:
|
||||
return tuple(value for key, transaction in batch for value in _row_params(table, key, transaction))
|
||||
|
||||
|
||||
def build_bulk_upsert(
|
||||
table: DailySpendTable,
|
||||
batch: Sequence[tuple[tuple[str, ...], SpendRow]],
|
||||
) -> tuple[str, tuple[SqlValue, ...]]:
|
||||
"""The single statement writing one merged batch, plus its positional arguments."""
|
||||
return _upsert_statement(table, batch, first_param=1), _params(table, batch)
|
||||
|
||||
|
||||
def build_bulk_upsert_with_global_rollup(
|
||||
table: DailySpendTable,
|
||||
batch: Sequence[tuple[tuple[str, ...], SpendRow]],
|
||||
) -> tuple[str, tuple[SqlValue, ...]]:
|
||||
"""One statement writing a batch to its table and, atomically, its key-free rollup
|
||||
to ``LiteLLM_DailyGlobalSpend``.
|
||||
|
||||
A data-modifying CTE runs both inserts in the same snapshot and transaction, so a
|
||||
batch that lands in one table lands in both and a retried deadlock replays both.
|
||||
Postgres does not order the CTE against the main statement, so two writers can still
|
||||
deadlock across the tables; the caller's deadlock retry covers that, and each insert
|
||||
takes its own rows in key order so same-table lock order stays deterministic.
|
||||
"""
|
||||
global_batch: Final = merge_by_conflict_key(GLOBAL_SPEND_TABLE, tuple(row for _, row in batch))
|
||||
entity_params: Final = _params(table, batch)
|
||||
sql: Final = (
|
||||
f"WITH entity_rows AS (\n{_upsert_statement(table, batch, first_param=1)}\nRETURNING 1)\n"
|
||||
f"{_upsert_statement(GLOBAL_SPEND_TABLE, global_batch, first_param=len(entity_params) + 1)}"
|
||||
)
|
||||
return sql, (*entity_params, *_params(GLOBAL_SPEND_TABLE, global_batch))
|
||||
|
|
|
|||
|
|
@ -46,6 +46,7 @@ from litellm.proxy._types import (
|
|||
from litellm.proxy.db.daily_spend_bulk_upsert import (
|
||||
DAILY_SPEND_TABLES,
|
||||
build_bulk_upsert,
|
||||
build_bulk_upsert_with_global_rollup,
|
||||
merge_by_conflict_key,
|
||||
)
|
||||
from litellm.proxy.db.db_transaction_queue.daily_spend_update_queue import (
|
||||
|
|
@ -1939,7 +1940,11 @@ class DBSpendUpdateWriter:
|
|||
merged_batch = merge_by_conflict_key(
|
||||
table=table, transactions=tuple(transactions_to_process.values())
|
||||
)
|
||||
sql, params = build_bulk_upsert(table=table, batch=merged_batch)
|
||||
sql, params = (
|
||||
build_bulk_upsert_with_global_rollup(table=table, batch=merged_batch)
|
||||
if entity_type == "user"
|
||||
else build_bulk_upsert(table=table, batch=merged_batch)
|
||||
)
|
||||
await prisma_client.db.execute_raw(sql, *params)
|
||||
except Exception as batch_error:
|
||||
# Log detailed error information for debugging batch upsert failures
|
||||
|
|
|
|||
|
|
@ -11,6 +11,8 @@ from typing_extensions import ReadOnly, TypedDict
|
|||
from litellm._logging import verbose_proxy_logger
|
||||
from litellm.constants import PTU_SENTINEL_API_KEY, USAGE_TOP_API_KEYS_LIMIT
|
||||
from litellm.proxy._types import CommonProxyErrors
|
||||
from litellm.proxy.db.daily_spend_bulk_upsert import GLOBAL_SPEND_TABLE
|
||||
from litellm.proxy.spend_tracking.daily_global_spend_rollup import reconciled_through
|
||||
from litellm.proxy.spend_tracking.key_metadata_recovery import (
|
||||
attach_user_emails,
|
||||
recover_double_hashed_key_metadata,
|
||||
|
|
@ -734,6 +736,30 @@ def _rollup_metric_select(table_name: str) -> str:
|
|||
_MODEL_GROUP_EXPR: Final = "COALESCE(NULLIF(model_group, ''), model)"
|
||||
|
||||
|
||||
async def key_free_source_table(prisma_client: PrismaClient, query: _AggregatedQueryKwargs) -> str | None:
|
||||
"""The table the key-free arm reads from, when the global rollup can answer instead of the per-key table.
|
||||
|
||||
Only an unfiltered read of the user table has the same rows as ``LiteLLM_DailyGlobalSpend``,
|
||||
and only through the day the reconcile marker has reached: the writer keeps that day
|
||||
current, later days are covered once the next run advances the marker.
|
||||
"""
|
||||
if query["table_name"] != "litellm_dailyuserspend":
|
||||
return None
|
||||
if query["entity_id"] is not None or query["api_key"] is not None or query["exclude_entity_ids"]:
|
||||
return None
|
||||
_, adjusted_end = _adjust_dates_for_timezone(
|
||||
query["start_date"], query["end_date"], query["timezone_offset_minutes"], query["include_current_utc_day"]
|
||||
)
|
||||
try:
|
||||
marker: Final = await reconciled_through(prisma_client)
|
||||
except Exception as exc: # noqa: BLE001 # the per-key table is always a correct answer, so never fail the read
|
||||
verbose_proxy_logger.warning("Could not read the daily global spend marker, using the per-key table: %s", exc)
|
||||
return None
|
||||
if marker is None or adjusted_end > marker:
|
||||
return None
|
||||
return GLOBAL_SPEND_TABLE.name
|
||||
|
||||
|
||||
def _build_aggregated_sql_query(
|
||||
*,
|
||||
table_name: str,
|
||||
|
|
@ -746,13 +772,16 @@ def _build_aggregated_sql_query(
|
|||
exclude_entity_ids: list[str] | None = None, # mutable-ok: filter union shared with the paginated path
|
||||
timezone_offset_minutes: int | None = None,
|
||||
include_current_utc_day: bool = False,
|
||||
key_free_table: str | None = None,
|
||||
) -> tuple[str, list[str]]: # mutable-ok: SQL text plus its ordered $N params
|
||||
"""Build the GROUPING SETS query for aggregated daily activity.
|
||||
|
||||
One statement, two UNION ALL arms over the same WHERE clause. The first arm is
|
||||
key-free: grand total, per-date totals and the (date, model / model_group /
|
||||
provider / mcp / endpoint) rollups, so its row count never grows with the number
|
||||
of keys. The second arm emits the (date, <dimension>, api_key) rollups for the
|
||||
of keys; it reads ``key_free_table`` when given (the global rollup, whose row count
|
||||
never grew with the number of keys to begin with) and the entity table otherwise.
|
||||
The second arm emits the (date, <dimension>, api_key) rollups for the
|
||||
USAGE_TOP_API_KEYS_LIMIT highest-spend keys only. Both arms share the 7-bit
|
||||
group_level bitmask (date, api_key, model, model_group, provider, mcp, endpoint).
|
||||
|
||||
|
|
@ -778,6 +807,7 @@ def _build_aggregated_sql_query(
|
|||
)
|
||||
sentinel_param: Final = f"${len(where_params) + 1}"
|
||||
metric_select: Final = _rollup_metric_select(table_name)
|
||||
key_free_source: Final = key_free_table or pg_table
|
||||
|
||||
# TODO: drop the successful_requests/failed_requests aggregates (and the
|
||||
# total_successful_requests metadata they feed) once the admin UI reads SGR
|
||||
|
|
@ -796,7 +826,7 @@ def _build_aggregated_sql_query(
|
|||
| GROUPING(model, {_MODEL_GROUP_EXPR},
|
||||
custom_llm_provider, mcp_namespaced_tool_name,
|
||||
endpoint) AS group_level,{metric_select}
|
||||
FROM "{pg_table}"
|
||||
FROM "{key_free_source}"
|
||||
WHERE {where_clause}
|
||||
GROUP BY GROUPING SETS (
|
||||
(date),
|
||||
|
|
@ -1387,7 +1417,9 @@ async def get_daily_activity_aggregated(
|
|||
timezone_offset_minutes=timezone_offset_minutes,
|
||||
include_current_utc_day=include_current_utc_day,
|
||||
)
|
||||
sql_query, sql_params = _build_aggregated_sql_query(**query_kwargs)
|
||||
sql_query, sql_params = _build_aggregated_sql_query(
|
||||
**query_kwargs, key_free_table=await key_free_source_table(prisma_client, query_kwargs)
|
||||
)
|
||||
entity_query: Final = _build_entity_rollup_sql_query(**query_kwargs) if include_entity_breakdown else None
|
||||
|
||||
raw_rows, raw_entity_rows = await asyncio.gather(
|
||||
|
|
|
|||
|
|
@ -259,6 +259,7 @@ from litellm.constants import (
|
|||
APSCHEDULER_MISFIRE_GRACE_TIME,
|
||||
APSCHEDULER_REPLACE_EXISTING,
|
||||
CLI_SSO_SESSION_TTL_SECONDS,
|
||||
DAILY_GLOBAL_SPEND_RECONCILE_JOB_ID,
|
||||
DAYS_IN_A_MONTH,
|
||||
DEFAULT_HEALTH_CHECK_INTERVAL,
|
||||
DEFAULT_MODEL_CREATED_AT_TIME,
|
||||
|
|
@ -662,6 +663,9 @@ from litellm.proxy.route_priority import hot_routes_first
|
|||
from litellm.proxy.search_endpoints.endpoints import router as search_router
|
||||
from litellm.proxy.shutdown.graceful_shutdown_manager import GracefulShutdownManager
|
||||
from litellm.proxy.spend_tracking.budget_reservation import get_budget_window_start
|
||||
from litellm.proxy.spend_tracking.daily_global_spend_rollup import (
|
||||
run_scheduled_daily_global_spend_reconcile,
|
||||
)
|
||||
from litellm.proxy.spend_tracking.spend_counter_batch import (
|
||||
PendingSpendIncrement,
|
||||
active_spend_counter_batch,
|
||||
|
|
@ -9970,6 +9974,12 @@ class ProxyStartupEvent:
|
|||
|
||||
await cls._initialize_spend_tracking_background_jobs(scheduler=scheduler)
|
||||
|
||||
cls._initialize_daily_global_spend_reconcile_job(
|
||||
scheduler=scheduler,
|
||||
proxy_logging_obj=proxy_logging_obj,
|
||||
prisma_client=prisma_client,
|
||||
)
|
||||
|
||||
### PTU DAILY ROLLUP ###
|
||||
from litellm.proxy.spend_tracking.ptu_feature_flag import (
|
||||
is_ptu_cost_attribution_enabled,
|
||||
|
|
@ -10311,6 +10321,39 @@ class ProxyStartupEvent:
|
|||
"LITELLM_EXPIRED_UI_SESSION_KEY_CLEANUP_ENABLED=true to enable)"
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _initialize_daily_global_spend_reconcile_job(
|
||||
cls,
|
||||
scheduler: AsyncIOScheduler,
|
||||
proxy_logging_obj: ProxyLogging,
|
||||
prisma_client: PrismaClient,
|
||||
) -> None:
|
||||
async def alert(message: str) -> None:
|
||||
await proxy_logging_obj.alerting_handler(
|
||||
message=message,
|
||||
level="High",
|
||||
alert_type=AlertType.failed_tracking_spend,
|
||||
)
|
||||
|
||||
async def reconcile() -> None:
|
||||
await run_scheduled_daily_global_spend_reconcile(
|
||||
prisma_client,
|
||||
pod_lock_manager=proxy_logging_obj.db_spend_update_writer.pod_lock_manager,
|
||||
alert=alert,
|
||||
)
|
||||
|
||||
scheduler.add_job(
|
||||
reconcile,
|
||||
"cron",
|
||||
hour=0,
|
||||
minute=30,
|
||||
timezone="UTC",
|
||||
id=DAILY_GLOBAL_SPEND_RECONCILE_JOB_ID,
|
||||
replace_existing=True,
|
||||
misfire_grace_time=APSCHEDULER_MISFIRE_GRACE_TIME,
|
||||
next_run_time=datetime.now(timezone.utc) + timedelta(minutes=2),
|
||||
)
|
||||
|
||||
@classmethod
|
||||
async def _initialize_slack_alerting_jobs(
|
||||
cls,
|
||||
|
|
|
|||
|
|
@ -809,6 +809,35 @@ model LiteLLM_DailyUserSpend {
|
|||
@@index([endpoint])
|
||||
}
|
||||
|
||||
// Key-free daily rollup of LiteLLM_DailyUserSpend, read by the global usage view
|
||||
model LiteLLM_DailyGlobalSpend {
|
||||
id String @id @default(uuid())
|
||||
date String
|
||||
model String?
|
||||
model_group String?
|
||||
custom_llm_provider String?
|
||||
mcp_namespaced_tool_name String?
|
||||
endpoint String?
|
||||
prompt_tokens BigInt @default(0)
|
||||
completion_tokens BigInt @default(0)
|
||||
cache_read_input_tokens BigInt @default(0)
|
||||
cache_creation_input_tokens BigInt @default(0)
|
||||
compression_saved_tokens BigInt @default(0)
|
||||
compression_savings_spend Float @default(0.0)
|
||||
prompt_caching_savings_spend Float @default(0.0)
|
||||
gateway_injected_caching_savings_spend Float @default(0.0)
|
||||
autorouter_savings_spend Float @default(0.0)
|
||||
spend Float @default(0.0)
|
||||
api_requests BigInt @default(0)
|
||||
successful_requests BigInt @default(0)
|
||||
failed_requests BigInt @default(0)
|
||||
created_at DateTime @default(now())
|
||||
updated_at DateTime @updatedAt
|
||||
|
||||
@@unique([date, model, model_group, custom_llm_provider, mcp_namespaced_tool_name, endpoint])
|
||||
@@index([date])
|
||||
}
|
||||
|
||||
// Track daily organization spend metrics per model and key
|
||||
model LiteLLM_DailyOrganizationSpend {
|
||||
id String @id @default(uuid())
|
||||
|
|
|
|||
235
litellm/proxy/spend_tracking/daily_global_spend_rollup.py
Normal file
235
litellm/proxy/spend_tracking/daily_global_spend_rollup.py
Normal file
|
|
@ -0,0 +1,235 @@
|
|||
"""Reconcile ``LiteLLM_DailyGlobalSpend`` from ``LiteLLM_DailyUserSpend``, one day per transaction.
|
||||
|
||||
The spend writer keeps both tables in step from the moment it is deployed; this job rolls up
|
||||
the days before that and records how far it has reached in ``LiteLLM_Config`` so usage reads
|
||||
know when the global table can answer for a date range. It runs as a background cron, never
|
||||
in a Prisma migration, since on a large deployment the aggregate is minutes of work.
|
||||
"""
|
||||
|
||||
from collections.abc import Awaitable, Callable
|
||||
from dataclasses import dataclass
|
||||
from datetime import date, datetime, timedelta, timezone
|
||||
from typing import TYPE_CHECKING, Final
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, ValidationError
|
||||
|
||||
from litellm._logging import verbose_proxy_logger
|
||||
from litellm.constants import (
|
||||
DAILY_GLOBAL_SPEND_RECONCILE_JOB_ID,
|
||||
DAILY_GLOBAL_SPEND_RECONCILE_LOCK_TTL_SECONDS,
|
||||
DAILY_GLOBAL_SPEND_RECONCILED_THROUGH_PARAM,
|
||||
)
|
||||
from litellm.proxy.db.daily_spend_bulk_upsert import GLOBAL_SPEND_TABLE
|
||||
from litellm.repositories.config_repository import ConfigRepository
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from litellm.caching.redis_cache import RedisCache
|
||||
from litellm.proxy.db.db_transaction_queue.pod_lock_manager import PodLockManager
|
||||
from litellm.proxy.utils import PrismaClient
|
||||
|
||||
_DAY_TRANSACTION_TIMEOUT: Final = timedelta(minutes=10)
|
||||
_REPLAY_DAYS: Final = 1
|
||||
_METRIC_COLUMNS: Final = (
|
||||
"prompt_tokens",
|
||||
"completion_tokens",
|
||||
"cache_read_input_tokens",
|
||||
"cache_creation_input_tokens",
|
||||
"compression_saved_tokens",
|
||||
"api_requests",
|
||||
"successful_requests",
|
||||
"failed_requests",
|
||||
"compression_savings_spend",
|
||||
"prompt_caching_savings_spend",
|
||||
"gateway_injected_caching_savings_spend",
|
||||
"autorouter_savings_spend",
|
||||
"spend",
|
||||
)
|
||||
|
||||
|
||||
def _quoted(columns: tuple[str, ...]) -> str:
|
||||
return ", ".join(f'"{column}"' for column in columns)
|
||||
|
||||
|
||||
def _reconcile_day_sql() -> str:
|
||||
key_columns: Final = GLOBAL_SPEND_TABLE.key_columns
|
||||
normalized_keys: Final = ", ".join(f"COALESCE(\"{column}\", '')" for column in key_columns)
|
||||
sums: Final = ", ".join(f'SUM("{column}")' for column in _METRIC_COLUMNS)
|
||||
overwrite: Final = ", ".join(f'"{column}" = EXCLUDED."{column}"' for column in _METRIC_COLUMNS)
|
||||
return (
|
||||
f'INSERT INTO "{GLOBAL_SPEND_TABLE.name}" ("id", {_quoted(key_columns)}, {_quoted(_METRIC_COLUMNS)}, '
|
||||
'"updated_at")\n'
|
||||
f"SELECT gen_random_uuid()::text, {normalized_keys}, {sums}, (NOW() AT TIME ZONE 'UTC')\n"
|
||||
'FROM "LiteLLM_DailyUserSpend" WHERE "date" = $1\n'
|
||||
f"GROUP BY {normalized_keys}\n"
|
||||
f"ON CONFLICT ({_quoted(key_columns)}) DO UPDATE SET {overwrite}, "
|
||||
"\"updated_at\" = (NOW() AT TIME ZONE 'UTC')"
|
||||
)
|
||||
|
||||
|
||||
RECONCILE_DAY_SQL: Final = _reconcile_day_sql()
|
||||
_LOCK_GLOBAL_TABLE_SQL: Final = f'LOCK TABLE "{GLOBAL_SPEND_TABLE.name}" IN EXCLUSIVE MODE'
|
||||
_PENDING_DAYS_SQL: Final = (
|
||||
'SELECT DISTINCT "date" FROM "LiteLLM_DailyUserSpend" WHERE "date" >= $1 AND "date" <= $2 ORDER BY "date"'
|
||||
)
|
||||
|
||||
|
||||
class ReconciledThrough(BaseModel):
|
||||
model_config = ConfigDict(frozen=True, extra="ignore")
|
||||
|
||||
reconciled_through: str
|
||||
|
||||
|
||||
class _MarkerRow(BaseModel):
|
||||
model_config = ConfigDict(frozen=True, extra="ignore", from_attributes=True)
|
||||
|
||||
param_value: object = None
|
||||
|
||||
|
||||
class _DateRow(BaseModel):
|
||||
model_config = ConfigDict(frozen=True, extra="ignore")
|
||||
|
||||
date: str
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ReconcileResult:
|
||||
days_reconciled: tuple[str, ...]
|
||||
reconciled_through: str | None
|
||||
failed_day: str | None = None
|
||||
|
||||
|
||||
def _marker_from_param_value(value: object) -> str | None:
|
||||
try:
|
||||
parsed: Final = (
|
||||
ReconciledThrough.model_validate_json(value)
|
||||
if isinstance(value, str)
|
||||
else ReconciledThrough.model_validate(value)
|
||||
)
|
||||
except ValidationError:
|
||||
return None
|
||||
return parsed.reconciled_through
|
||||
|
||||
|
||||
async def reconciled_through(prisma_client: "PrismaClient") -> str | None:
|
||||
"""The last UTC day ``LiteLLM_DailyGlobalSpend`` is known to cover, or None before the first run."""
|
||||
from litellm.proxy.utils import get_config_param
|
||||
|
||||
row: Final = await get_config_param(prisma_client, DAILY_GLOBAL_SPEND_RECONCILED_THROUGH_PARAM)
|
||||
return None if row is None else _marker_from_param_value(_MarkerRow.model_validate(row).param_value)
|
||||
|
||||
|
||||
async def _record_reconciled_through(prisma_client: "PrismaClient", day: str) -> None:
|
||||
from litellm.proxy.utils import invalidate_config_param
|
||||
|
||||
await ConfigRepository(prisma_client).set_param(
|
||||
DAILY_GLOBAL_SPEND_RECONCILED_THROUGH_PARAM, ReconciledThrough(reconciled_through=day).model_dump_json()
|
||||
)
|
||||
await invalidate_config_param(DAILY_GLOBAL_SPEND_RECONCILED_THROUGH_PARAM)
|
||||
|
||||
|
||||
def _first_pending_day(marker: str | None) -> str:
|
||||
if marker is None:
|
||||
return ""
|
||||
return (date.fromisoformat(marker) - timedelta(days=_REPLAY_DAYS)).isoformat()
|
||||
|
||||
|
||||
async def pending_days(prisma_client: "PrismaClient", today: date) -> tuple[str, ...]:
|
||||
"""Every UTC day through today still to roll up, oldest first; the marker day and the one
|
||||
before it are replayed so rows flushed by a pre-writer pod during a rolling deploy are folded in."""
|
||||
marker: Final = await reconciled_through(prisma_client)
|
||||
rows: Final = await prisma_client.db.query_raw(_PENDING_DAYS_SQL, _first_pending_day(marker), today.isoformat())
|
||||
return tuple(sorted({*(_DateRow.model_validate(row).date for row in rows), today.isoformat()}))
|
||||
|
||||
|
||||
async def reconcile_day(prisma_client: "PrismaClient", day: str) -> None:
|
||||
"""Rewrite one day of the global table from the per-key sums; the table lock keeps the
|
||||
writer's increments out between the aggregate and the overwrite so none are lost."""
|
||||
async with prisma_client.db.tx(timeout=_DAY_TRANSACTION_TIMEOUT) as transaction:
|
||||
await transaction.execute_raw(_LOCK_GLOBAL_TABLE_SQL)
|
||||
await transaction.execute_raw(RECONCILE_DAY_SQL, day)
|
||||
|
||||
|
||||
async def run_daily_global_spend_reconcile(
|
||||
prisma_client: "PrismaClient",
|
||||
today: date | None = None,
|
||||
) -> ReconcileResult:
|
||||
"""Roll up every pending day, advancing the marker after each; a failing day stops the run
|
||||
with the marker on the last good day so the next run resumes there."""
|
||||
effective_today: Final = today or datetime.now(timezone.utc).date()
|
||||
days: Final = await pending_days(prisma_client, effective_today)
|
||||
done: Final = await _reconcile_until_failure(prisma_client, days)
|
||||
failed: Final = days[len(done)] if len(done) < len(days) else None
|
||||
marker: Final = done[-1] if done else await reconciled_through(prisma_client)
|
||||
return ReconcileResult(days_reconciled=done, reconciled_through=marker, failed_day=failed)
|
||||
|
||||
|
||||
async def _reconcile_until_failure(prisma_client: "PrismaClient", days: tuple[str, ...]) -> tuple[str, ...]:
|
||||
for index, day in enumerate(days):
|
||||
if not await _reconcile_and_record(prisma_client, day):
|
||||
return days[:index]
|
||||
return days
|
||||
|
||||
|
||||
async def _reconcile_and_record(prisma_client: "PrismaClient", day: str) -> bool:
|
||||
try:
|
||||
await reconcile_day(prisma_client, day)
|
||||
await _record_reconciled_through(prisma_client, day)
|
||||
except Exception as exc: # noqa: BLE001 # one bad day must not lose the days already done
|
||||
verbose_proxy_logger.exception("Daily global spend reconcile: day %s failed: %s", day, exc)
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
async def run_scheduled_daily_global_spend_reconcile(
|
||||
prisma_client: "PrismaClient",
|
||||
pod_lock_manager: "PodLockManager | None" = None,
|
||||
alert: Callable[[str], Awaitable[None]] | None = None,
|
||||
today: date | None = None,
|
||||
) -> ReconcileResult | None:
|
||||
"""Run the reconcile under a cross-pod lock so one proxy does the work; the lock only saves
|
||||
effort (each day is an idempotent rewrite), so an unreachable Redis runs unguarded rather than skipping."""
|
||||
redis_cache: Final = None if pod_lock_manager is None else pod_lock_manager.redis_cache
|
||||
if pod_lock_manager is None or redis_cache is None:
|
||||
return await _run_and_alert(prisma_client, alert=alert, today=today)
|
||||
|
||||
acquired: Final = await pod_lock_manager.acquire_lock(
|
||||
cronjob_id=DAILY_GLOBAL_SPEND_RECONCILE_JOB_ID, ttl=DAILY_GLOBAL_SPEND_RECONCILE_LOCK_TTL_SECONDS
|
||||
)
|
||||
if not acquired and await _lock_is_held(pod_lock_manager, redis_cache):
|
||||
verbose_proxy_logger.info("Daily global spend reconcile: another pod holds the lock, skipping this run")
|
||||
return None
|
||||
try:
|
||||
return await _run_and_alert(prisma_client, alert=alert, today=today)
|
||||
finally:
|
||||
if acquired:
|
||||
await pod_lock_manager.release_lock(cronjob_id=DAILY_GLOBAL_SPEND_RECONCILE_JOB_ID)
|
||||
|
||||
|
||||
async def _lock_is_held(pod_lock_manager: "PodLockManager", redis_cache: "RedisCache") -> bool:
|
||||
try:
|
||||
lock_key: Final = pod_lock_manager.get_redis_lock_key(DAILY_GLOBAL_SPEND_RECONCILE_JOB_ID)
|
||||
return bool(await redis_cache.async_get_cache(lock_key))
|
||||
except Exception as exc: # noqa: BLE001 # an unreadable lock must not skip the run
|
||||
verbose_proxy_logger.warning("Daily global spend reconcile: could not read the lock: %s", exc)
|
||||
return False
|
||||
|
||||
|
||||
async def _run_and_alert(
|
||||
prisma_client: "PrismaClient",
|
||||
*,
|
||||
alert: Callable[[str], Awaitable[None]] | None,
|
||||
today: date | None,
|
||||
) -> ReconcileResult:
|
||||
result: Final = await run_daily_global_spend_reconcile(prisma_client, today=today)
|
||||
if result.days_reconciled:
|
||||
verbose_proxy_logger.info(
|
||||
"Daily global spend reconcile: rolled up %d day(s), reconciled through %s",
|
||||
len(result.days_reconciled),
|
||||
result.reconciled_through,
|
||||
)
|
||||
if result.failed_day is not None and alert is not None:
|
||||
await alert(
|
||||
f"Daily global spend reconcile stopped at {result.failed_day}; usage totals keep reading the per-key "
|
||||
f"table for ranges past {result.reconciled_through or 'the beginning'} until the next run succeeds."
|
||||
)
|
||||
return result
|
||||
|
|
@ -809,6 +809,35 @@ model LiteLLM_DailyUserSpend {
|
|||
@@index([endpoint])
|
||||
}
|
||||
|
||||
// Key-free daily rollup of LiteLLM_DailyUserSpend, read by the global usage view
|
||||
model LiteLLM_DailyGlobalSpend {
|
||||
id String @id @default(uuid())
|
||||
date String
|
||||
model String?
|
||||
model_group String?
|
||||
custom_llm_provider String?
|
||||
mcp_namespaced_tool_name String?
|
||||
endpoint String?
|
||||
prompt_tokens BigInt @default(0)
|
||||
completion_tokens BigInt @default(0)
|
||||
cache_read_input_tokens BigInt @default(0)
|
||||
cache_creation_input_tokens BigInt @default(0)
|
||||
compression_saved_tokens BigInt @default(0)
|
||||
compression_savings_spend Float @default(0.0)
|
||||
prompt_caching_savings_spend Float @default(0.0)
|
||||
gateway_injected_caching_savings_spend Float @default(0.0)
|
||||
autorouter_savings_spend Float @default(0.0)
|
||||
spend Float @default(0.0)
|
||||
api_requests BigInt @default(0)
|
||||
successful_requests BigInt @default(0)
|
||||
failed_requests BigInt @default(0)
|
||||
created_at DateTime @default(now())
|
||||
updated_at DateTime @updatedAt
|
||||
|
||||
@@unique([date, model, model_group, custom_llm_provider, mcp_namespaced_tool_name, endpoint])
|
||||
@@index([date])
|
||||
}
|
||||
|
||||
// Track daily organization spend metrics per model and key
|
||||
model LiteLLM_DailyOrganizationSpend {
|
||||
id String @id @default(uuid())
|
||||
|
|
|
|||
|
|
@ -1,12 +1,19 @@
|
|||
"""Tests for the single-statement daily spend upsert (LIT-5291)."""
|
||||
|
||||
import pathlib
|
||||
import re
|
||||
from typing import Final
|
||||
|
||||
import psycopg
|
||||
import pytest
|
||||
from psycopg.rows import dict_row
|
||||
from pytest_postgresql import factories
|
||||
|
||||
from litellm.proxy.db.daily_spend_bulk_upsert import (
|
||||
DAILY_SPEND_TABLES,
|
||||
GLOBAL_SPEND_TABLE,
|
||||
build_bulk_upsert,
|
||||
build_bulk_upsert_with_global_rollup,
|
||||
conflict_key,
|
||||
merge_by_conflict_key,
|
||||
)
|
||||
|
|
@ -185,3 +192,146 @@ async def test_writer_survives_a_transaction_whose_key_columns_are_null():
|
|||
_, params = prisma_client.db.statements[0]
|
||||
assert None not in params[:9]
|
||||
assert transactions == {}
|
||||
|
||||
|
||||
def user_txn(**overrides):
|
||||
txn = {**tag_txn(), "user_id": "u-1", **overrides}
|
||||
del txn["tag"]
|
||||
del txn["request_id"]
|
||||
return txn
|
||||
|
||||
|
||||
def _bound_rows(insert_sql: str, params: tuple[object, ...]) -> list[dict[str, object]]:
|
||||
"""Each VALUES row of one INSERT as a column -> bound value mapping, consuming params in order."""
|
||||
header = re.search(r"INSERT INTO \"[A-Za-z_]+\" \(([^)]*)\)", insert_sql)
|
||||
assert header is not None, insert_sql
|
||||
columns = [c.strip('"') for c in header.group(1).split(", ") if c != '"updated_at"']
|
||||
row_count = insert_sql.split("ON CONFLICT", 1)[0].count("(NOW() AT TIME ZONE 'UTC'))")
|
||||
return [dict(zip(columns, params[i * len(columns) : (i + 1) * len(columns)])) for i in range(row_count)]
|
||||
|
||||
|
||||
def test_global_rollup_folds_every_key_and_user_into_one_row_per_dimension_tuple():
|
||||
"""The global table has no api_key or user_id, so a batch spread over many keys and
|
||||
users must collapse to one row per (date, model, group, provider, mcp, endpoint)."""
|
||||
batch = merge_by_conflict_key(
|
||||
USER_TABLE,
|
||||
tuple(user_txn(user_id=f"u-{i}", api_key=f"sk-{i}", spend=1.0, api_requests=1) for i in range(5))
|
||||
+ (user_txn(user_id="u-0", api_key="sk-0", model="claude", spend=10.0, api_requests=3),),
|
||||
)
|
||||
|
||||
sql, params = build_bulk_upsert_with_global_rollup(USER_TABLE, batch)
|
||||
|
||||
entity_insert, global_insert = sql.split("RETURNING 1)")
|
||||
entity_rows = _bound_rows(entity_insert, params)
|
||||
global_rows = _bound_rows(global_insert, params[len(entity_rows) * len(entity_rows[0]) :])
|
||||
assert len(entity_rows) == 6
|
||||
assert 'INSERT INTO "LiteLLM_DailyGlobalSpend"' in global_insert
|
||||
assert [(r["model"], r["spend"], r["api_requests"]) for r in global_rows] == [
|
||||
("claude", 10.0, 3),
|
||||
("gpt-4o-mini", 5.0, 5),
|
||||
]
|
||||
assert all("api_key" not in r and "user_id" not in r for r in global_rows)
|
||||
conflict = re.search(r"ON CONFLICT \(([^)]*)\)", global_insert)
|
||||
assert conflict is not None
|
||||
assert conflict.group(1) == ", ".join(f'"{c}"' for c in GLOBAL_SPEND_TABLE.key_columns)
|
||||
|
||||
|
||||
def test_global_rollup_params_follow_the_entity_params_in_one_placeholder_sequence():
|
||||
"""Both inserts bind from one flat tuple, so the global arm's placeholders must start
|
||||
exactly where the entity arm's stop or every value lands one column off."""
|
||||
batch = merge_by_conflict_key(USER_TABLE, (user_txn(),))
|
||||
|
||||
sql, params = build_bulk_upsert_with_global_rollup(USER_TABLE, batch)
|
||||
|
||||
placeholders = [int(n) for n in re.findall(r"\$(\d+)::", sql)]
|
||||
assert placeholders == list(range(1, len(params) + 1))
|
||||
|
||||
|
||||
_bulk_upsert_postgresql_proc: Final = factories.postgresql_proc()
|
||||
_bulk_upsert_postgresql: Final = factories.postgresql("_bulk_upsert_postgresql_proc")
|
||||
|
||||
_MIGRATIONS_DIR: Final = (
|
||||
pathlib.Path(__file__).resolve().parents[4] / "litellm-proxy-extras" / "litellm_proxy_extras" / "migrations"
|
||||
)
|
||||
_GLOBAL_SPEND_MIGRATION: Final = _MIGRATIONS_DIR / "20260915000000_add_daily_global_spend" / "migration.sql"
|
||||
|
||||
_DAILY_USER_SPEND_DDL: Final = """
|
||||
CREATE TABLE "LiteLLM_DailyUserSpend" (
|
||||
id TEXT PRIMARY KEY,
|
||||
user_id TEXT,
|
||||
date TEXT NOT NULL,
|
||||
api_key TEXT NOT NULL,
|
||||
model TEXT,
|
||||
model_group TEXT,
|
||||
custom_llm_provider TEXT,
|
||||
mcp_namespaced_tool_name TEXT,
|
||||
endpoint TEXT,
|
||||
prompt_tokens BIGINT DEFAULT 0,
|
||||
completion_tokens BIGINT DEFAULT 0,
|
||||
cache_read_input_tokens BIGINT DEFAULT 0,
|
||||
cache_creation_input_tokens BIGINT DEFAULT 0,
|
||||
compression_saved_tokens BIGINT DEFAULT 0,
|
||||
compression_savings_spend DOUBLE PRECISION DEFAULT 0,
|
||||
prompt_caching_savings_spend DOUBLE PRECISION DEFAULT 0,
|
||||
gateway_injected_caching_savings_spend DOUBLE PRECISION DEFAULT 0,
|
||||
autorouter_savings_spend DOUBLE PRECISION DEFAULT 0,
|
||||
spend DOUBLE PRECISION DEFAULT 0,
|
||||
api_requests BIGINT DEFAULT 0,
|
||||
successful_requests BIGINT DEFAULT 0,
|
||||
failed_requests BIGINT DEFAULT 0,
|
||||
created_at TIMESTAMP DEFAULT now(),
|
||||
updated_at TIMESTAMP,
|
||||
UNIQUE (user_id, date, api_key, model, custom_llm_provider, mcp_namespaced_tool_name, endpoint)
|
||||
)
|
||||
"""
|
||||
|
||||
|
||||
def _execute_dollar_sql(conn: psycopg.Connection, sql: str, params: tuple[object, ...]) -> None:
|
||||
converted: Final = re.sub(r"\$(\d+)", r"%(p\1)s", sql)
|
||||
conn.execute(
|
||||
converted, # pyright: ignore[reportArgumentType] # psycopg stubs want a literal-typed query
|
||||
{f"p{i}": v for i, v in enumerate(params, start=1)},
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_global_rollup_equals_the_per_key_sums_after_repeated_flushes(_bulk_upsert_postgresql: psycopg.Connection):
|
||||
"""Against real Postgres and the shipped migration: two flushes of a mixed batch leave
|
||||
the global table exactly equal to the per-key table summed over user and key, with the
|
||||
NULL and '' spellings of a dimension folded into one row."""
|
||||
conn: Final = _bulk_upsert_postgresql
|
||||
conn.execute(_DAILY_USER_SPEND_DDL) # pyright: ignore[reportArgumentType] # DDL literal
|
||||
conn.execute(_GLOBAL_SPEND_MIGRATION.read_text()) # pyright: ignore[reportArgumentType] # DDL literal
|
||||
conn.commit()
|
||||
|
||||
batch = merge_by_conflict_key(
|
||||
USER_TABLE,
|
||||
(
|
||||
user_txn(user_id="u-1", api_key="sk-1", spend=1.0, prompt_tokens=10),
|
||||
user_txn(user_id="u-2", api_key="sk-2", spend=2.0, prompt_tokens=20),
|
||||
user_txn(user_id="u-1", api_key="sk-3", model=None, custom_llm_provider=None, spend=4.0),
|
||||
user_txn(user_id="u-3", api_key="sk-4", model="", custom_llm_provider="", spend=8.0),
|
||||
),
|
||||
)
|
||||
sql, params = build_bulk_upsert_with_global_rollup(USER_TABLE, batch)
|
||||
_execute_dollar_sql(conn, sql, params)
|
||||
_execute_dollar_sql(conn, sql, params)
|
||||
|
||||
with conn.cursor(row_factory=dict_row) as cur:
|
||||
global_rows = cur.execute(
|
||||
'SELECT model, spend, prompt_tokens, api_requests FROM "LiteLLM_DailyGlobalSpend" ORDER BY model'
|
||||
).fetchall()
|
||||
per_key = cur.execute(
|
||||
"""
|
||||
SELECT COALESCE(model, '') AS model, SUM(spend) AS spend, SUM(prompt_tokens) AS prompt_tokens,
|
||||
SUM(api_requests) AS api_requests
|
||||
FROM "LiteLLM_DailyUserSpend" GROUP BY COALESCE(model, '') ORDER BY 1
|
||||
"""
|
||||
).fetchall()
|
||||
|
||||
assert [row["model"] for row in global_rows] == ["", "gpt-4o-mini"]
|
||||
assert [(r["model"], r["spend"], int(r["prompt_tokens"]), int(r["api_requests"])) for r in global_rows] == [
|
||||
(r["model"], float(r["spend"]), int(r["prompt_tokens"]), int(r["api_requests"])) for r in per_key
|
||||
]
|
||||
assert global_rows[0]["spend"] == pytest.approx(24.0)
|
||||
assert global_rows[1]["spend"] == pytest.approx(6.0)
|
||||
|
|
|
|||
|
|
@ -254,14 +254,19 @@ class _RecordingPrisma:
|
|||
|
||||
|
||||
def _row_values(statement: Statement, column: str) -> list[object]:
|
||||
"""Every row's value for one column, read out of the flat parameter tuple."""
|
||||
"""Every row's value for one column of the first INSERT, read out of the flat parameter tuple.
|
||||
|
||||
The user-table statement chains a global rollup INSERT after its own, so the row count
|
||||
comes from the first INSERT's VALUES rather than from the parameter count.
|
||||
"""
|
||||
sql, params = statement
|
||||
header = re.search(r"INSERT INTO \"[A-Za-z_]+\" \(([^)]*)\)", sql)
|
||||
assert header is not None, sql
|
||||
columns = header.group(1).split(", ")
|
||||
stride = len(columns) - 1 # updated_at is inlined, not bound
|
||||
offset = columns.index(f'"{column}"')
|
||||
return [params[row * stride + offset] for row in range(len(params) // stride)]
|
||||
rows = sql.split("ON CONFLICT", 1)[0].count("(NOW() AT TIME ZONE 'UTC'))")
|
||||
return [params[row * stride + offset] for row in range(rows)]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
@ -1463,6 +1468,98 @@ async def test_update_daily_spend_keeps_failed_transactions_for_retry():
|
|||
assert daily_spend_transactions == expected
|
||||
|
||||
|
||||
def _entity_txn(entity_field: str, entity_id: str, api_key: str) -> dict[str, object]:
|
||||
txn = _daily_txn()
|
||||
del txn["user_id"]
|
||||
return {**txn, entity_field: entity_id, "api_key": api_key}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_user_flush_writes_the_global_rollup_in_the_same_statement():
|
||||
"""The user flush is the one place per-key spend becomes key-free spend, so a batch spread
|
||||
over many keys must land in LiteLLM_DailyGlobalSpend as one row in the same statement.
|
||||
A separate statement would let a crash between the two leave the tables out of sync."""
|
||||
prisma_client = _RecordingPrisma()
|
||||
txns = {f"k{i}": _entity_txn("user_id", f"user-{i}", f"sk-{i}") for i in range(4)}
|
||||
|
||||
await DBSpendUpdateWriter._update_daily_spend(
|
||||
n_retry_times=0,
|
||||
prisma_client=prisma_client,
|
||||
proxy_logging_obj=MagicMock(),
|
||||
daily_spend_transactions=txns,
|
||||
entity_type="user",
|
||||
entity_id_field="user_id",
|
||||
)
|
||||
|
||||
assert len(prisma_client.db.statements) == 1
|
||||
sql, params = prisma_client.db.statements[0]
|
||||
assert sql.count('INSERT INTO "LiteLLM_DailyUserSpend"') == 1
|
||||
assert sql.count('INSERT INTO "LiteLLM_DailyGlobalSpend"') == 1
|
||||
assert sql.index('"LiteLLM_DailyUserSpend"') < sql.index('"LiteLLM_DailyGlobalSpend"')
|
||||
global_insert = sql.split('INSERT INTO "LiteLLM_DailyGlobalSpend"', 1)[1]
|
||||
assert global_insert.split("ON CONFLICT", 1)[0].count("(NOW() AT TIME ZONE 'UTC'))") == 1
|
||||
assert "api_key" not in global_insert
|
||||
assert params.count(0.4) == 1
|
||||
assert txns == {}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
("entity_type", "entity_field"),
|
||||
[
|
||||
("team", "team_id"),
|
||||
("org", "organization_id"),
|
||||
("tag", "tag"),
|
||||
("end_user", "end_user_id"),
|
||||
("agent", "agent_id"),
|
||||
],
|
||||
)
|
||||
async def test_other_entity_flushes_leave_the_global_table_alone(entity_type, entity_field):
|
||||
"""Every entity table sees the same request, so writing the rollup from more than one of
|
||||
them would count each request once per entity type."""
|
||||
prisma_client = _RecordingPrisma()
|
||||
txn = _entity_txn(entity_field, "e-1", "sk-1")
|
||||
if entity_type == "tag":
|
||||
txn["request_id"] = "req-1"
|
||||
|
||||
await DBSpendUpdateWriter._update_daily_spend(
|
||||
n_retry_times=0,
|
||||
prisma_client=prisma_client,
|
||||
proxy_logging_obj=MagicMock(),
|
||||
daily_spend_transactions={"k": txn},
|
||||
entity_type=entity_type,
|
||||
entity_id_field=entity_field,
|
||||
)
|
||||
|
||||
(sql, _params) = prisma_client.db.statements[0]
|
||||
assert "LiteLLM_DailyGlobalSpend" not in sql
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_failed_chained_user_flush_keeps_every_transaction_for_retry():
|
||||
def raise_outage():
|
||||
raise ValueError("simulated database outage")
|
||||
|
||||
prisma_client = _RecordingPrisma(execute_raw=raise_outage)
|
||||
txns = {f"k{i}": _entity_txn("user_id", f"user-{i}", f"sk-{i}") for i in range(3)}
|
||||
expected = dict(txns)
|
||||
mock_proxy_logging = MagicMock()
|
||||
mock_proxy_logging.failure_handler = AsyncMock()
|
||||
|
||||
with pytest.raises(ValueError, match="simulated database outage"):
|
||||
await DBSpendUpdateWriter._update_daily_spend(
|
||||
n_retry_times=0,
|
||||
prisma_client=prisma_client,
|
||||
proxy_logging_obj=mock_proxy_logging,
|
||||
daily_spend_transactions=txns,
|
||||
entity_type="user",
|
||||
entity_id_field="user_id",
|
||||
)
|
||||
|
||||
assert txns == expected
|
||||
assert 'INSERT INTO "LiteLLM_DailyGlobalSpend"' in prisma_client.db.statements[0][0]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_commit_key_spend_updates_includes_last_active():
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -13,7 +13,13 @@ from pytest_postgresql import factories
|
|||
from litellm.proxy.spend_tracking.ptu_feature_flag import PTU_COST_ATTRIBUTION_ENV_VAR
|
||||
|
||||
|
||||
from litellm.constants import PTU_SENTINEL_API_KEY, USAGE_TOP_API_KEYS_LIMIT
|
||||
import pathlib
|
||||
|
||||
from litellm.constants import (
|
||||
DAILY_GLOBAL_SPEND_RECONCILED_THROUGH_PARAM,
|
||||
PTU_SENTINEL_API_KEY,
|
||||
USAGE_TOP_API_KEYS_LIMIT,
|
||||
)
|
||||
from litellm.proxy.management_endpoints.common_daily_activity import (
|
||||
_adjust_dates_for_timezone,
|
||||
_build_aggregated_sql_query,
|
||||
|
|
@ -23,8 +29,11 @@ from litellm.proxy.management_endpoints.common_daily_activity import (
|
|||
get_api_key_metadata,
|
||||
get_daily_activity,
|
||||
get_daily_activity_aggregated,
|
||||
key_free_source_table,
|
||||
update_metrics,
|
||||
)
|
||||
from litellm.proxy.spend_tracking.daily_global_spend_rollup import RECONCILE_DAY_SQL
|
||||
from litellm.proxy.utils import evict_config_param
|
||||
from litellm.types.proxy.management_endpoints.common_daily_activity import (
|
||||
DailySpendMetadata,
|
||||
SpendMetrics,
|
||||
|
|
@ -1618,6 +1627,149 @@ async def test_get_daily_activity_aggregated_explicit_api_key_filter_scopes_both
|
|||
assert set(day.breakdown.models["gpt-5"].api_key_breakdown) == {"key-1"}
|
||||
|
||||
|
||||
def _prisma_with_marker(marker: str | None) -> MagicMock:
|
||||
prisma = MagicMock()
|
||||
prisma.db = MagicMock()
|
||||
prisma.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[])
|
||||
prisma.db.litellm_deletedverificationtoken.find_many = AsyncMock(return_value=[])
|
||||
row = None if marker is None else SimpleNamespace(param_name="m", param_value=f'{{"reconciled_through": "{marker}"}}')
|
||||
prisma.get_generic_data = AsyncMock(return_value=row)
|
||||
return prisma
|
||||
|
||||
|
||||
def _unfiltered_user_query(**overrides):
|
||||
return {
|
||||
"table_name": "litellm_dailyuserspend",
|
||||
"entity_id_field": "user_id",
|
||||
"entity_id": None,
|
||||
"start_date": "2026-06-01",
|
||||
"end_date": "2026-06-02",
|
||||
"model": None,
|
||||
"api_key": None,
|
||||
"exclude_entity_ids": None,
|
||||
"timezone_offset_minutes": None,
|
||||
"include_current_utc_day": False,
|
||||
**overrides,
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
("marker", "overrides", "expected"),
|
||||
[
|
||||
("2026-06-02", {}, "LiteLLM_DailyGlobalSpend"),
|
||||
("2026-06-02", {"model": "gpt-5"}, "LiteLLM_DailyGlobalSpend"),
|
||||
("2026-06-01", {}, None),
|
||||
(None, {}, None),
|
||||
("2026-06-02", {"api_key": "sk-1"}, None),
|
||||
("2026-06-02", {"api_key": []}, None),
|
||||
("2026-06-02", {"entity_id": "u-1"}, None),
|
||||
("2026-06-02", {"exclude_entity_ids": ["u-1"]}, None),
|
||||
("2026-06-02", {"table_name": "litellm_dailyteamspend", "entity_id_field": "team_id"}, None),
|
||||
],
|
||||
)
|
||||
async def test_key_free_source_table_routes_only_unfiltered_user_reads_within_the_marker(marker, overrides, expected):
|
||||
"""Anything that filters by key or entity has no counterpart in the global table, and a
|
||||
range the reconcile has not reached must stay on the per-key table."""
|
||||
await evict_config_param(DAILY_GLOBAL_SPEND_RECONCILED_THROUGH_PARAM)
|
||||
prisma = _prisma_with_marker(marker)
|
||||
|
||||
assert await key_free_source_table(prisma, _unfiltered_user_query(**overrides)) == expected
|
||||
await evict_config_param(DAILY_GLOBAL_SPEND_RECONCILED_THROUGH_PARAM)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_key_free_source_table_judges_the_timezone_extended_end_not_the_requested_one():
|
||||
"""A caller west of UTC asking through their local today gets today's UTC bucket added to
|
||||
the range; the marker must cover that extended day, not just the requested end."""
|
||||
await evict_config_param(DAILY_GLOBAL_SPEND_RECONCILED_THROUGH_PARAM)
|
||||
today_utc: Final = datetime.now(timezone.utc).date()
|
||||
yesterday: Final = (today_utc - timedelta(days=1)).isoformat()
|
||||
query: Final = _unfiltered_user_query(
|
||||
start_date=yesterday, end_date=yesterday, timezone_offset_minutes=24 * 60, include_current_utc_day=True
|
||||
)
|
||||
|
||||
assert await key_free_source_table(_prisma_with_marker(yesterday), query) is None
|
||||
await evict_config_param(DAILY_GLOBAL_SPEND_RECONCILED_THROUGH_PARAM)
|
||||
assert await key_free_source_table(_prisma_with_marker(today_utc.isoformat()), query) == "LiteLLM_DailyGlobalSpend"
|
||||
await evict_config_param(DAILY_GLOBAL_SPEND_RECONCILED_THROUGH_PARAM)
|
||||
|
||||
|
||||
_GLOBAL_SPEND_MIGRATION: Final = (
|
||||
pathlib.Path(__file__).resolve().parents[4]
|
||||
/ "litellm-proxy-extras"
|
||||
/ "litellm_proxy_extras"
|
||||
/ "migrations"
|
||||
/ "20260915000000_add_daily_global_spend"
|
||||
/ "migration.sql"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_daily_activity_aggregated_reads_the_global_table_for_the_key_free_arm(
|
||||
_aggregated_postgresql: psycopg.Connection,
|
||||
):
|
||||
"""With the range reconciled, the key-free arm reads LiteLLM_DailyGlobalSpend while the
|
||||
per-key arm stays on the user table, and the response is identical to the all-per-key
|
||||
read: same totals, same rollups, same top keys."""
|
||||
n_keys: Final = USAGE_TOP_API_KEYS_LIMIT + 3
|
||||
rows: Final = [
|
||||
(
|
||||
f"row-{day}-{i:03d}",
|
||||
f"user-{i % 7}",
|
||||
day,
|
||||
f"key-{i:03d}",
|
||||
"gpt-5" if i % 2 else "claude",
|
||||
"" if i % 3 else "gpt-5",
|
||||
"openai" if i % 2 else None,
|
||||
"/v1/chat/completions" if i % 5 else None,
|
||||
10,
|
||||
float(i + 1),
|
||||
1,
|
||||
1,
|
||||
)
|
||||
for day in ("2026-06-01", "2026-06-02")
|
||||
for i in range(n_keys)
|
||||
]
|
||||
_seed_daily_user_spend(_aggregated_postgresql, rows)
|
||||
with _aggregated_postgresql.cursor() as cur:
|
||||
cur.execute(_GLOBAL_SPEND_MIGRATION.read_text()) # pyright: ignore[reportArgumentType] # DDL literal
|
||||
for day in ("2026-06-01", "2026-06-02"):
|
||||
cur.execute(
|
||||
re.sub(r"\$(\d+)", r"%(p\1)s", RECONCILE_DAY_SQL), # pyright: ignore[reportArgumentType] # $N -> psycopg
|
||||
{"p1": day},
|
||||
)
|
||||
_aggregated_postgresql.commit()
|
||||
|
||||
async def read(marker: str | None, sql_seen: list[str]):
|
||||
await evict_config_param(DAILY_GLOBAL_SPEND_RECONCILED_THROUGH_PARAM)
|
||||
prisma = _prisma_with_marker(marker)
|
||||
run_query = _psycopg_query_raw(_aggregated_postgresql, [])
|
||||
|
||||
async def query_raw(sql: str, *params: str):
|
||||
sql_seen.append(sql)
|
||||
return await run_query(sql, *params)
|
||||
|
||||
prisma.db.query_raw = query_raw
|
||||
return await get_daily_activity_aggregated(
|
||||
prisma_client=prisma,
|
||||
entity_metadata_field=None,
|
||||
**_unfiltered_user_query(),
|
||||
)
|
||||
|
||||
per_key_sql: Final[list[str]] = [] # mutable-ok: out-param for the query_raw shim
|
||||
global_sql: Final[list[str]] = [] # mutable-ok: out-param for the query_raw shim
|
||||
from_per_key = await read(None, per_key_sql)
|
||||
from_global = await read("2026-06-02", global_sql)
|
||||
await evict_config_param(DAILY_GLOBAL_SPEND_RECONCILED_THROUGH_PARAM)
|
||||
|
||||
assert per_key_sql[0].count('FROM "LiteLLM_DailyGlobalSpend"') == 0
|
||||
assert global_sql[0].count('FROM "LiteLLM_DailyGlobalSpend"') == 1
|
||||
assert global_sql[0].count('FROM "LiteLLM_DailyUserSpend"') == 2
|
||||
assert from_global.model_dump() == from_per_key.model_dump()
|
||||
assert from_global.metadata.total_spend == pytest.approx(2 * sum(float(i + 1) for i in range(n_keys)))
|
||||
assert len(from_global.results[0].breakdown.api_keys) == USAGE_TOP_API_KEYS_LIMIT
|
||||
assert set(from_global.results[0].breakdown.model_groups) == {"gpt-5", "claude"}
|
||||
def _no_spend_record():
|
||||
"""A rollup row for a key with no spend, where SUM() returns NULL (None)."""
|
||||
return SimpleNamespace(
|
||||
|
|
|
|||
|
|
@ -1042,6 +1042,54 @@ async def test_spend_report_locks_are_never_released():
|
|||
proxy_logging_obj.db_spend_update_writer.pod_lock_manager.release_lock.assert_not_awaited()
|
||||
|
||||
|
||||
def _init_daily_global_spend_reconcile_job() -> tuple[MagicMock, MagicMock, MagicMock]:
|
||||
scheduler = MagicMock()
|
||||
proxy_logging_obj = MagicMock()
|
||||
proxy_logging_obj.alerting_handler = AsyncMock()
|
||||
prisma_client = MagicMock()
|
||||
ProxyStartupEvent._initialize_daily_global_spend_reconcile_job(
|
||||
scheduler=scheduler,
|
||||
proxy_logging_obj=proxy_logging_obj,
|
||||
prisma_client=prisma_client,
|
||||
)
|
||||
return scheduler, proxy_logging_obj, prisma_client
|
||||
|
||||
|
||||
def test_daily_global_spend_reconcile_job_is_scheduled_nightly_with_an_immediate_catch_up_run():
|
||||
"""Startup schedules the LiteLLM_DailyGlobalSpend backfill a couple of minutes out, so a
|
||||
fresh deploy switches usage reads to the global table without waiting for the nightly
|
||||
run, and replaces any previous registration of the same job id."""
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from litellm.constants import DAILY_GLOBAL_SPEND_RECONCILE_JOB_ID
|
||||
|
||||
scheduler, _, _ = _init_daily_global_spend_reconcile_job()
|
||||
|
||||
(call,) = scheduler.add_job.call_args_list
|
||||
assert call.kwargs["id"] == DAILY_GLOBAL_SPEND_RECONCILE_JOB_ID
|
||||
assert call.kwargs["replace_existing"] is True
|
||||
assert call.args[1:] == ("cron",)
|
||||
assert (call.kwargs["hour"], call.kwargs["minute"], call.kwargs["timezone"]) == (0, 30, "UTC")
|
||||
assert timedelta(0) < call.kwargs["next_run_time"] - datetime.now(timezone.utc) <= timedelta(minutes=2)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_daily_global_spend_reconcile_job_runs_under_the_pod_lock_and_alerts_through_the_proxy(monkeypatch):
|
||||
scheduler, proxy_logging_obj, prisma_client = _init_daily_global_spend_reconcile_job()
|
||||
run = AsyncMock()
|
||||
monkeypatch.setattr(ps, "run_scheduled_daily_global_spend_reconcile", run)
|
||||
|
||||
await scheduler.add_job.call_args.args[0]()
|
||||
|
||||
run.assert_awaited_once()
|
||||
assert run.await_args.args == (prisma_client,)
|
||||
assert run.await_args.kwargs["pod_lock_manager"] is proxy_logging_obj.db_spend_update_writer.pod_lock_manager
|
||||
await run.await_args.kwargs["alert"]("day 2026-09-01 failed")
|
||||
proxy_logging_obj.alerting_handler.assert_awaited_once()
|
||||
assert proxy_logging_obj.alerting_handler.await_args.kwargs["message"] == "day 2026-09-01 failed"
|
||||
assert proxy_logging_obj.alerting_handler.await_args.kwargs["level"] == "High"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_prometheus_fallback_stats_job_skipped_when_another_pod_holds_the_lock(monkeypatch):
|
||||
"""The boot-time send goes through the same gate, so a losing pod sends nothing at all:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,382 @@
|
|||
"""Tests for the LiteLLM_DailyGlobalSpend reconcile job (LIT-7818)."""
|
||||
|
||||
import pathlib
|
||||
import re
|
||||
from contextlib import asynccontextmanager
|
||||
from datetime import date
|
||||
from typing import Final
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import psycopg
|
||||
import pytest
|
||||
from psycopg.rows import dict_row
|
||||
from pytest_postgresql import factories
|
||||
|
||||
from litellm.constants import DAILY_GLOBAL_SPEND_RECONCILED_THROUGH_PARAM
|
||||
from litellm.proxy.db.daily_spend_bulk_upsert import (
|
||||
DAILY_SPEND_TABLES,
|
||||
build_bulk_upsert_with_global_rollup,
|
||||
merge_by_conflict_key,
|
||||
)
|
||||
from litellm.proxy.spend_tracking.daily_global_spend_rollup import (
|
||||
RECONCILE_DAY_SQL,
|
||||
reconciled_through,
|
||||
run_daily_global_spend_reconcile,
|
||||
run_scheduled_daily_global_spend_reconcile,
|
||||
)
|
||||
from litellm.proxy.utils import evict_config_param
|
||||
|
||||
USER_TABLE: Final = DAILY_SPEND_TABLES["user"]
|
||||
TODAY: Final = date(2026, 9, 15)
|
||||
|
||||
|
||||
class _FakeConfigRow:
|
||||
def __init__(self, param_name: str, param_value: object) -> None:
|
||||
self.param_name = param_name
|
||||
self.param_value = param_value
|
||||
|
||||
|
||||
class _FakeConfigTable:
|
||||
def __init__(self) -> None:
|
||||
self.rows: dict[str, object] = {}
|
||||
|
||||
async def upsert(self, *, where: dict[str, str], data: dict[str, dict[str, str]]) -> _FakeConfigRow:
|
||||
self.rows[where["param_name"]] = data["update"]["param_value"]
|
||||
return _FakeConfigRow(where["param_name"], data["update"]["param_value"])
|
||||
|
||||
|
||||
class _FakeTransaction:
|
||||
def __init__(self, prisma: "_FakePrisma") -> None:
|
||||
self._prisma = prisma
|
||||
|
||||
async def execute_raw(self, sql: str, *params: str) -> int:
|
||||
if "LOCK TABLE" in sql:
|
||||
self._prisma.locks_taken += 1
|
||||
return 0
|
||||
(day,) = params
|
||||
if day in self._prisma.failing_days:
|
||||
raise RuntimeError(f"day {day} exploded")
|
||||
self._prisma.reconciled.append(day)
|
||||
return 1
|
||||
|
||||
|
||||
class _FakeDb:
|
||||
def __init__(self, prisma: "_FakePrisma") -> None:
|
||||
self._prisma = prisma
|
||||
self.litellm_config = _FakeConfigTable()
|
||||
|
||||
async def query_raw(self, sql: str, *params: str) -> list[dict[str, str]]:
|
||||
first, last = params
|
||||
return [{"date": d} for d in sorted(self._prisma.user_days) if first <= d <= last]
|
||||
|
||||
@asynccontextmanager
|
||||
async def tx(self, timeout: object):
|
||||
yield _FakeTransaction(self._prisma)
|
||||
|
||||
|
||||
class _FakePrisma:
|
||||
"""Enough of PrismaClient for the reconcile: per-key dates, a config table, and a transaction."""
|
||||
|
||||
def __init__(self, user_days: tuple[str, ...], failing_days: frozenset[str] = frozenset()) -> None:
|
||||
self.user_days = user_days
|
||||
self.failing_days = failing_days
|
||||
self.reconciled: list[str] = []
|
||||
self.locks_taken = 0
|
||||
self.db = _FakeDb(self)
|
||||
|
||||
async def get_generic_data(self, key: str, value: str, table_name: str) -> _FakeConfigRow | None:
|
||||
stored = self.db.litellm_config.rows.get(value)
|
||||
return None if stored is None else _FakeConfigRow(value, stored)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
async def _fresh_marker_cache():
|
||||
await evict_config_param(DAILY_GLOBAL_SPEND_RECONCILED_THROUGH_PARAM)
|
||||
yield
|
||||
await evict_config_param(DAILY_GLOBAL_SPEND_RECONCILED_THROUGH_PARAM)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_first_run_rolls_up_every_historical_day_and_today_then_marks_today():
|
||||
"""Before any marker exists, every day with per-key rows is rolled up, plus today even
|
||||
with no rows yet, so reads for ranges ending today can switch to the global table."""
|
||||
prisma = _FakePrisma(user_days=("2026-09-01", "2026-09-03", "2026-09-14"))
|
||||
|
||||
result = await run_daily_global_spend_reconcile(prisma, today=TODAY)
|
||||
|
||||
assert result.days_reconciled == ("2026-09-01", "2026-09-03", "2026-09-14", "2026-09-15")
|
||||
assert result.failed_day is None
|
||||
assert result.reconciled_through == "2026-09-15"
|
||||
assert await reconciled_through(prisma) == "2026-09-15"
|
||||
assert prisma.locks_taken == 4
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_later_run_replays_the_marker_day_and_the_day_before_only():
|
||||
"""Days older than marker-1 are settled; the marker day and its predecessor are replayed so
|
||||
rows a pre-writer pod flushed around midnight during a rolling deploy get folded in."""
|
||||
prisma = _FakePrisma(user_days=("2026-09-01", "2026-09-12", "2026-09-13", "2026-09-14"))
|
||||
await run_daily_global_spend_reconcile(prisma, today=date(2026, 9, 13))
|
||||
prisma.reconciled.clear()
|
||||
|
||||
result = await run_daily_global_spend_reconcile(prisma, today=TODAY)
|
||||
|
||||
assert result.days_reconciled == ("2026-09-12", "2026-09-13", "2026-09-14", "2026-09-15")
|
||||
assert "2026-09-01" not in prisma.reconciled
|
||||
assert await reconciled_through(prisma) == "2026-09-15"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_failing_day_stops_the_run_and_leaves_the_marker_on_the_last_good_day():
|
||||
"""The marker may never claim a day that was not rewritten: reads past it would then trust
|
||||
a global table missing that day's spend."""
|
||||
prisma = _FakePrisma(user_days=("2026-09-01", "2026-09-02", "2026-09-03"), failing_days=frozenset({"2026-09-02"}))
|
||||
|
||||
result = await run_daily_global_spend_reconcile(prisma, today=TODAY)
|
||||
|
||||
assert result.days_reconciled == ("2026-09-01",)
|
||||
assert result.failed_day == "2026-09-02"
|
||||
assert result.reconciled_through == "2026-09-01"
|
||||
assert prisma.reconciled == ["2026-09-01"]
|
||||
assert await reconciled_through(prisma) == "2026-09-01"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_the_next_run_resumes_from_the_failed_day():
|
||||
prisma = _FakePrisma(user_days=("2026-09-01", "2026-09-02", "2026-09-03"), failing_days=frozenset({"2026-09-02"}))
|
||||
await run_daily_global_spend_reconcile(prisma, today=TODAY)
|
||||
prisma.failing_days = frozenset()
|
||||
|
||||
result = await run_daily_global_spend_reconcile(prisma, today=TODAY)
|
||||
|
||||
assert result.days_reconciled == ("2026-09-01", "2026-09-02", "2026-09-03", "2026-09-15")
|
||||
assert await reconciled_through(prisma) == "2026-09-15"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_failure_with_nothing_done_reports_the_previous_marker_and_alerts():
|
||||
"""A pre-writer pod flushing rows for the day before the marker is exactly the replay case;
|
||||
when that replay fails the marker must stay put and the operator must hear about it."""
|
||||
prisma = _FakePrisma(user_days=("2026-09-13",))
|
||||
await run_daily_global_spend_reconcile(prisma, today=date(2026, 9, 13))
|
||||
prisma.user_days = ("2026-09-12", "2026-09-13")
|
||||
prisma.failing_days = frozenset({"2026-09-12"})
|
||||
alert = AsyncMock()
|
||||
|
||||
result = await run_scheduled_daily_global_spend_reconcile(prisma, pod_lock_manager=None, alert=alert, today=TODAY)
|
||||
|
||||
assert result is not None
|
||||
assert result.days_reconciled == ()
|
||||
assert result.failed_day == "2026-09-12"
|
||||
assert result.reconciled_through == "2026-09-13"
|
||||
alert.assert_awaited_once()
|
||||
assert "2026-09-12" in alert.await_args.args[0]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_clean_run_does_not_alert():
|
||||
prisma = _FakePrisma(user_days=("2026-09-13",))
|
||||
alert = AsyncMock()
|
||||
|
||||
await run_scheduled_daily_global_spend_reconcile(prisma, pod_lock_manager=None, alert=alert, today=TODAY)
|
||||
|
||||
alert.assert_not_awaited()
|
||||
|
||||
|
||||
def _pod_lock(acquired: bool) -> MagicMock:
|
||||
lock = MagicMock()
|
||||
lock.redis_cache = MagicMock()
|
||||
lock.redis_cache.async_get_cache = AsyncMock(return_value="other-pod")
|
||||
lock.get_redis_lock_key = MagicMock(return_value="lock-key")
|
||||
lock.acquire_lock = AsyncMock(return_value=acquired)
|
||||
lock.release_lock = AsyncMock()
|
||||
return lock
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_scheduled_run_skips_when_another_pod_holds_the_lock():
|
||||
prisma = _FakePrisma(user_days=("2026-09-13",))
|
||||
lock = _pod_lock(acquired=False)
|
||||
|
||||
result = await run_scheduled_daily_global_spend_reconcile(prisma, pod_lock_manager=lock, today=TODAY)
|
||||
|
||||
assert result is None
|
||||
assert prisma.reconciled == []
|
||||
lock.release_lock.assert_not_awaited()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_scheduled_run_runs_and_releases_the_lock_when_it_wins():
|
||||
prisma = _FakePrisma(user_days=("2026-09-13",))
|
||||
lock = _pod_lock(acquired=True)
|
||||
|
||||
result = await run_scheduled_daily_global_spend_reconcile(prisma, pod_lock_manager=lock, today=TODAY)
|
||||
|
||||
assert result is not None and result.days_reconciled == ("2026-09-13", "2026-09-15")
|
||||
lock.release_lock.assert_awaited_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_scheduled_run_proceeds_when_the_lock_cannot_be_acquired_or_read():
|
||||
"""A Redis outage must not stall the backfill: the day rewrite is idempotent, so running
|
||||
twice is only wasted effort while skipping forever leaves usage on the slow path."""
|
||||
prisma = _FakePrisma(user_days=("2026-09-13",))
|
||||
lock = _pod_lock(acquired=False)
|
||||
lock.redis_cache.async_get_cache = AsyncMock(side_effect=ConnectionError("redis down"))
|
||||
|
||||
result = await run_scheduled_daily_global_spend_reconcile(prisma, pod_lock_manager=lock, today=TODAY)
|
||||
|
||||
assert result is not None and result.days_reconciled == ("2026-09-13", "2026-09-15")
|
||||
lock.release_lock.assert_not_awaited()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_marker_is_read_back_from_the_json_string_the_config_table_stores():
|
||||
prisma = _FakePrisma(user_days=())
|
||||
prisma.db.litellm_config.rows[DAILY_GLOBAL_SPEND_RECONCILED_THROUGH_PARAM] = '{"reconciled_through": "2026-09-10"}'
|
||||
|
||||
assert await reconciled_through(prisma) == "2026-09-10"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_unparseable_marker_reads_as_never_reconciled():
|
||||
prisma = _FakePrisma(user_days=())
|
||||
prisma.db.litellm_config.rows[DAILY_GLOBAL_SPEND_RECONCILED_THROUGH_PARAM] = '{"something_else": 1}'
|
||||
|
||||
assert await reconciled_through(prisma) is None
|
||||
|
||||
|
||||
_rollup_postgresql_proc: Final = factories.postgresql_proc()
|
||||
_rollup_postgresql: Final = factories.postgresql("_rollup_postgresql_proc")
|
||||
|
||||
_MIGRATIONS_DIR: Final = (
|
||||
pathlib.Path(__file__).resolve().parents[4] / "litellm-proxy-extras" / "litellm_proxy_extras" / "migrations"
|
||||
)
|
||||
_GLOBAL_SPEND_MIGRATION: Final = _MIGRATIONS_DIR / "20260915000000_add_daily_global_spend" / "migration.sql"
|
||||
|
||||
_DAILY_USER_SPEND_DDL: Final = """
|
||||
CREATE TABLE "LiteLLM_DailyUserSpend" (
|
||||
id TEXT PRIMARY KEY,
|
||||
user_id TEXT,
|
||||
date TEXT NOT NULL,
|
||||
api_key TEXT NOT NULL,
|
||||
model TEXT,
|
||||
model_group TEXT,
|
||||
custom_llm_provider TEXT,
|
||||
mcp_namespaced_tool_name TEXT,
|
||||
endpoint TEXT,
|
||||
prompt_tokens BIGINT DEFAULT 0,
|
||||
completion_tokens BIGINT DEFAULT 0,
|
||||
cache_read_input_tokens BIGINT DEFAULT 0,
|
||||
cache_creation_input_tokens BIGINT DEFAULT 0,
|
||||
compression_saved_tokens BIGINT DEFAULT 0,
|
||||
compression_savings_spend DOUBLE PRECISION DEFAULT 0,
|
||||
prompt_caching_savings_spend DOUBLE PRECISION DEFAULT 0,
|
||||
gateway_injected_caching_savings_spend DOUBLE PRECISION DEFAULT 0,
|
||||
autorouter_savings_spend DOUBLE PRECISION DEFAULT 0,
|
||||
spend DOUBLE PRECISION DEFAULT 0,
|
||||
api_requests BIGINT DEFAULT 0,
|
||||
successful_requests BIGINT DEFAULT 0,
|
||||
failed_requests BIGINT DEFAULT 0,
|
||||
created_at TIMESTAMP DEFAULT now(),
|
||||
updated_at TIMESTAMP,
|
||||
UNIQUE (user_id, date, api_key, model, custom_llm_provider, mcp_namespaced_tool_name, endpoint)
|
||||
)
|
||||
"""
|
||||
|
||||
_PER_KEY_SUMS_SQL: Final = """
|
||||
SELECT COALESCE(model, '') AS model, COALESCE(model_group, '') AS model_group,
|
||||
COALESCE(custom_llm_provider, '') AS custom_llm_provider,
|
||||
SUM(spend) AS spend, SUM(prompt_tokens) AS prompt_tokens, SUM(api_requests) AS api_requests
|
||||
FROM "LiteLLM_DailyUserSpend" WHERE date = %s
|
||||
GROUP BY 1, 2, 3 ORDER BY 1, 2, 3
|
||||
"""
|
||||
_GLOBAL_ROWS_SQL: Final = """
|
||||
SELECT model, model_group, custom_llm_provider, spend, prompt_tokens, api_requests
|
||||
FROM "LiteLLM_DailyGlobalSpend" WHERE date = %s ORDER BY 1, 2, 3
|
||||
"""
|
||||
|
||||
|
||||
def _execute_dollar_sql(conn: psycopg.Connection, sql: str, params: tuple[object, ...]) -> None:
|
||||
converted: Final = re.sub(r"\$(\d+)", r"%(p\1)s", sql)
|
||||
conn.execute(
|
||||
converted, # pyright: ignore[reportArgumentType] # psycopg stubs want a literal-typed query
|
||||
{f"p{i}": v for i, v in enumerate(params, start=1)},
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
|
||||
def _user_txn(**overrides):
|
||||
return {
|
||||
"user_id": "u-1",
|
||||
"date": "2026-09-14",
|
||||
"api_key": "sk-1",
|
||||
"model": "gpt-5",
|
||||
"model_group": "gpt-5",
|
||||
"custom_llm_provider": "openai",
|
||||
"mcp_namespaced_tool_name": "",
|
||||
"endpoint": "/chat/completions",
|
||||
"prompt_tokens": 10,
|
||||
"completion_tokens": 20,
|
||||
"spend": 1.0,
|
||||
"api_requests": 1,
|
||||
"successful_requests": 1,
|
||||
"failed_requests": 0,
|
||||
**overrides,
|
||||
}
|
||||
|
||||
|
||||
def _normalized(rows: list[dict[str, object]]) -> list[tuple[object, ...]]:
|
||||
return [
|
||||
(
|
||||
r["model"],
|
||||
r["model_group"],
|
||||
r["custom_llm_provider"],
|
||||
float(r["spend"]),
|
||||
int(r["prompt_tokens"]),
|
||||
int(r["api_requests"]),
|
||||
) # pyright: ignore[reportArgumentType] # dict_row values are untyped
|
||||
for r in rows
|
||||
]
|
||||
|
||||
|
||||
def test_reconcile_day_sql_makes_the_global_day_equal_the_per_key_sums(_rollup_postgresql: psycopg.Connection):
|
||||
"""Against real Postgres and the shipped migration: rows the writer never saw (a
|
||||
pre-writer pod's flush, NULL and '' dimension spellings) end up folded into the global
|
||||
day, running the day twice changes nothing, and other days are left alone."""
|
||||
conn: Final = _rollup_postgresql
|
||||
conn.execute(_DAILY_USER_SPEND_DDL) # pyright: ignore[reportArgumentType] # DDL literal
|
||||
conn.execute(_GLOBAL_SPEND_MIGRATION.read_text()) # pyright: ignore[reportArgumentType] # DDL literal
|
||||
conn.commit()
|
||||
|
||||
written_batch = merge_by_conflict_key(
|
||||
USER_TABLE,
|
||||
(_user_txn(api_key="sk-1", spend=1.0), _user_txn(api_key="sk-2", user_id="u-2", spend=2.0, prompt_tokens=20)),
|
||||
)
|
||||
_execute_dollar_sql(conn, *build_bulk_upsert_with_global_rollup(USER_TABLE, written_batch))
|
||||
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO "LiteLLM_DailyUserSpend"
|
||||
(id, user_id, date, api_key, model, model_group, custom_llm_provider, mcp_namespaced_tool_name,
|
||||
endpoint, prompt_tokens, spend, api_requests)
|
||||
VALUES
|
||||
('legacy-1', 'u-9', '2026-09-14', 'sk-9', 'gpt-5', NULL, 'openai', NULL, NULL, 5, 4.0, 1),
|
||||
('legacy-2', 'u-9', '2026-09-14', 'sk-9', 'gpt-5', '', 'openai', '', '', 5, 8.0, 1),
|
||||
('legacy-3', 'u-9', '2026-09-13', 'sk-9', 'claude', '', 'anthropic', '', '', 7, 16.0, 1)
|
||||
"""
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
_execute_dollar_sql(conn, RECONCILE_DAY_SQL, ("2026-09-14",))
|
||||
_execute_dollar_sql(conn, RECONCILE_DAY_SQL, ("2026-09-14",))
|
||||
|
||||
with conn.cursor(row_factory=dict_row) as cur:
|
||||
global_rows = cur.execute(_GLOBAL_ROWS_SQL, ("2026-09-14",)).fetchall()
|
||||
per_key = cur.execute(_PER_KEY_SUMS_SQL, ("2026-09-14",)).fetchall()
|
||||
untouched = cur.execute(_GLOBAL_ROWS_SQL, ("2026-09-13",)).fetchall()
|
||||
|
||||
assert _normalized(global_rows) == _normalized(per_key)
|
||||
assert sum(float(r["spend"]) for r in global_rows) == pytest.approx(15.0) # pyright: ignore[reportArgumentType] # dict_row values are untyped
|
||||
assert [(r["model"], r["model_group"]) for r in global_rows] == [("gpt-5", ""), ("gpt-5", "gpt-5")]
|
||||
assert untouched == []
|
||||
Loading…
Add table
Reference in a new issue