Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
c8a2d8c
feat(proxy): add LiteLLM_DailyGlobalSpend key-free rollup for the usa…
yassin-berriai Sep 15, 2026
ad8de0e
fix(proxy): roll up only closed days into LiteLLM_DailyGlobalSpend an…
yassin-berriai Sep 15, 2026
84c098d
fix(proxy): fold late-arriving per-key spend into already rolled-up g…
yassin-berriai Sep 16, 2026
225fc53
Merge branch 'litellm_usage_key_free_aggregate_split' into litellm_da…
yassin-berriai Sep 16, 2026
0601d2b
feat(proxy): carry response time metrics through LiteLLM_DailyGlobalS…
yassin-berriai Sep 16, 2026
7c5fa53
Merge remote-tracking branch 'origin/litellm_usage_key_free_aggregate…
yassin-berriai Sep 16, 2026
834313a
test(proxy): assert the global rollup split and scheduler through beh…
yassin-berriai Sep 16, 2026
f25d659
fix(proxy): take the closed-day cutoff for the global spend rollup fr…
yassin-berriai Sep 16, 2026
3b2b9bf
Merge branch 'litellm_usage_key_free_aggregate_split' into litellm_da…
yassin-berriai Sep 17, 2026
f631301
Merge branch 'litellm_usage_key_free_aggregate_split' into litellm_da…
yassin-berriai Sep 18, 2026
b92820d
Merge branch 'litellm_usage_key_free_aggregate_split' into litellm_da…
yassin-berriai Sep 18, 2026
6a6ae2d
Merge remote-tracking branch 'origin/main' into litellm_daily_global_…
yassin-berriai Sep 18, 2026
abf530f
fix(proxy): never rewind the daily global spend marker from an overla…
yassin-berriai Sep 18, 2026
3449ae9
fix(proxy): advance the daily global spend marker in one conditional …
yassin-berriai Sep 18, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -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");
29 changes: 29 additions & 0 deletions litellm-proxy-extras/litellm_proxy_extras/schema.prisma
Original file line number Diff line number Diff line change
Expand Up @@ -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())
Expand Down
3 changes: 3 additions & 0 deletions litellm/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -2034,6 +2034,9 @@
# 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
Expand Down
96 changes: 72 additions & 24 deletions litellm/proxy/db/daily_spend_bulk_upsert.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,28 +25,40 @@

@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


# 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": 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),
"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),
}
)

# 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")
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",
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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 ()),
Expand All @@ -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'))"
Expand All @@ -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))
7 changes: 6 additions & 1 deletion litellm/proxy/db/db_spend_update_writer.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@
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 (
Expand Down Expand Up @@ -1939,7 +1940,11 @@ async def _update_daily_spend(
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
Expand Down
38 changes: 35 additions & 3 deletions litellm/proxy/management_endpoints/common_daily_activity.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,8 @@
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,
Expand Down Expand Up @@ -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
Comment thread
greptile-apps[bot] marked this conversation as resolved.
Resolved
_, 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,
Expand All @@ -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).

Expand All @@ -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
Expand All @@ -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),
Expand Down Expand Up @@ -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(
Expand Down
Loading
Loading