mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-07 08:26:10 +00:00
feat(proxy): add warning/error level logging throughout spend tracking lifecycle
Elevate silent debug-level and bare except:pass error paths to warning/error so spend tracking failures are visible in production logs. All new log messages are prefixed with "Spend tracking -" for easy filtering. Changes cover the full request-to-DB lifecycle: enqueue, in-memory flush, Redis buffer push/pop, DB commit, cache updates, spend log writes, and pod lock management. Also fixes a copy-paste bug in _update_team_cache that logged "end user" instead of "team".
This commit is contained in:
parent
ac2f2da0d7
commit
e6640cc622
7 changed files with 177 additions and 35 deletions
|
|
@ -167,8 +167,16 @@ class DBSpendUpdateWriter:
|
|||
|
||||
verbose_proxy_logger.debug("Runs spend update on all tables")
|
||||
except Exception:
|
||||
verbose_proxy_logger.debug(
|
||||
f"Error updating Prisma database: {traceback.format_exc()}"
|
||||
verbose_proxy_logger.error(
|
||||
"Spend tracking - update_database failed. All spend updates for this request will be lost. "
|
||||
"response_cost=%s, token=%s, user_id=%s, team_id=%s, org_id=%s, end_user_id=%s - %s",
|
||||
response_cost,
|
||||
token,
|
||||
user_id,
|
||||
team_id,
|
||||
org_id,
|
||||
end_user_id,
|
||||
traceback.format_exc(),
|
||||
)
|
||||
|
||||
async def _batch_database_updates(
|
||||
|
|
@ -389,9 +397,14 @@ class DBSpendUpdateWriter:
|
|||
)
|
||||
)
|
||||
except Exception as e:
|
||||
verbose_proxy_logger.debug(
|
||||
"\033[91m"
|
||||
+ f"Update User DB call failed to execute {str(e)}\n{traceback.format_exc()}"
|
||||
verbose_proxy_logger.error(
|
||||
"Spend tracking - failed to enqueue user spend update. "
|
||||
"user_id=%s, end_user_id=%s, response_cost=%s - %s\n%s",
|
||||
user_id,
|
||||
end_user_id,
|
||||
response_cost,
|
||||
str(e),
|
||||
traceback.format_exc(),
|
||||
)
|
||||
|
||||
async def _update_team_db(
|
||||
|
|
@ -428,11 +441,23 @@ class DBSpendUpdateWriter:
|
|||
response_cost=response_cost,
|
||||
)
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
except Exception as e:
|
||||
verbose_proxy_logger.error(
|
||||
"Spend tracking - failed to enqueue team member spend update. "
|
||||
"team_id=%s, user_id=%s, response_cost=%s - %s",
|
||||
team_id,
|
||||
user_id,
|
||||
response_cost,
|
||||
str(e),
|
||||
)
|
||||
except Exception as e:
|
||||
verbose_proxy_logger.debug(
|
||||
f"Update Team DB failed to execute - {str(e)}\n{traceback.format_exc()}"
|
||||
verbose_proxy_logger.error(
|
||||
"Spend tracking - failed to enqueue team spend update. "
|
||||
"team_id=%s, response_cost=%s - %s\n%s",
|
||||
team_id,
|
||||
response_cost,
|
||||
str(e),
|
||||
traceback.format_exc(),
|
||||
)
|
||||
raise e
|
||||
|
||||
|
|
@ -457,8 +482,13 @@ class DBSpendUpdateWriter:
|
|||
)
|
||||
)
|
||||
except Exception as e:
|
||||
verbose_proxy_logger.debug(
|
||||
f"Update Org DB failed to execute - {str(e)}\n{traceback.format_exc()}"
|
||||
verbose_proxy_logger.error(
|
||||
"Spend tracking - failed to enqueue org spend update. "
|
||||
"org_id=%s, response_cost=%s - %s\n%s",
|
||||
org_id,
|
||||
response_cost,
|
||||
str(e),
|
||||
traceback.format_exc(),
|
||||
)
|
||||
raise e
|
||||
|
||||
|
|
@ -505,8 +535,13 @@ class DBSpendUpdateWriter:
|
|||
)
|
||||
)
|
||||
except Exception as e:
|
||||
verbose_proxy_logger.debug(
|
||||
f"Update Tag DB failed to execute - {str(e)}\n{traceback.format_exc()}"
|
||||
verbose_proxy_logger.error(
|
||||
"Spend tracking - failed to enqueue tag spend update. "
|
||||
"request_tags=%s, response_cost=%s - %s\n%s",
|
||||
request_tags,
|
||||
response_cost,
|
||||
str(e),
|
||||
traceback.format_exc(),
|
||||
)
|
||||
raise e
|
||||
|
||||
|
|
@ -607,6 +642,17 @@ class DBSpendUpdateWriter:
|
|||
await self.redis_update_buffer.get_all_update_transactions_from_redis_buffer()
|
||||
)
|
||||
if db_spend_update_transactions is not None:
|
||||
verbose_proxy_logger.info(
|
||||
"Spend tracking - committing spend updates from Redis to DB: "
|
||||
"keys=%d, users=%d, teams=%d, orgs=%d, end_users=%d, team_members=%d, tags=%d",
|
||||
len(db_spend_update_transactions.get("key_list_transactions") or {}),
|
||||
len(db_spend_update_transactions.get("user_list_transactions") or {}),
|
||||
len(db_spend_update_transactions.get("team_list_transactions") or {}),
|
||||
len(db_spend_update_transactions.get("org_list_transactions") or {}),
|
||||
len(db_spend_update_transactions.get("end_user_list_transactions") or {}),
|
||||
len(db_spend_update_transactions.get("team_member_list_transactions") or {}),
|
||||
len(db_spend_update_transactions.get("tag_list_transactions") or {}),
|
||||
)
|
||||
await self._commit_spend_updates_to_db(
|
||||
prisma_client=prisma_client,
|
||||
n_retry_times=n_retry_times,
|
||||
|
|
@ -677,7 +723,12 @@ class DBSpendUpdateWriter:
|
|||
daily_spend_transactions=daily_agent_spend_update_transactions,
|
||||
)
|
||||
except Exception as e:
|
||||
verbose_proxy_logger.error(f"Error committing spend updates: {e}")
|
||||
verbose_proxy_logger.error(
|
||||
"Spend tracking - failed to commit spend updates from Redis to DB. "
|
||||
"Data already popped from Redis may be lost. Error: %s\n%s",
|
||||
str(e),
|
||||
traceback.format_exc(),
|
||||
)
|
||||
finally:
|
||||
await self.pod_lock_manager.release_lock(
|
||||
cronjob_id=DB_SPEND_UPDATE_JOB_NAME,
|
||||
|
|
@ -702,6 +753,22 @@ class DBSpendUpdateWriter:
|
|||
db_spend_update_transactions = (
|
||||
await self.spend_update_queue.flush_and_get_aggregated_db_spend_update_transactions()
|
||||
)
|
||||
if any(
|
||||
len(v) > 0
|
||||
for v in db_spend_update_transactions.values()
|
||||
if isinstance(v, dict)
|
||||
):
|
||||
verbose_proxy_logger.info(
|
||||
"Spend tracking - committing spend updates to DB (no Redis buffer): "
|
||||
"keys=%d, users=%d, teams=%d, orgs=%d, end_users=%d, team_members=%d, tags=%d",
|
||||
len(db_spend_update_transactions.get("key_list_transactions") or {}),
|
||||
len(db_spend_update_transactions.get("user_list_transactions") or {}),
|
||||
len(db_spend_update_transactions.get("team_list_transactions") or {}),
|
||||
len(db_spend_update_transactions.get("org_list_transactions") or {}),
|
||||
len(db_spend_update_transactions.get("end_user_list_transactions") or {}),
|
||||
len(db_spend_update_transactions.get("team_member_list_transactions") or {}),
|
||||
len(db_spend_update_transactions.get("tag_list_transactions") or {}),
|
||||
)
|
||||
await self._commit_spend_updates_to_db(
|
||||
prisma_client=prisma_client,
|
||||
n_retry_times=n_retry_times,
|
||||
|
|
|
|||
|
|
@ -86,6 +86,11 @@ class DailySpendUpdateQueue(BaseUpdateQueue):
|
|||
) -> Dict[str, BaseDailySpendTransaction]:
|
||||
"""Get all updates from the queue and return all updates aggregated by daily_transaction_key. Works for both user and team spend updates."""
|
||||
updates = await self.flush_all_updates_from_in_memory_queue()
|
||||
if len(updates) > 0:
|
||||
verbose_proxy_logger.info(
|
||||
"Spend tracking - flushed %d daily spend update items from in-memory queue",
|
||||
len(updates),
|
||||
)
|
||||
aggregated_daily_spend_update_transactions = (
|
||||
DailySpendUpdateQueue.get_aggregated_daily_spend_update_transactions(
|
||||
updates
|
||||
|
|
|
|||
|
|
@ -80,6 +80,14 @@ class PodLockManager:
|
|||
)
|
||||
self._emit_acquired_lock_event(cronjob_id, self.pod_id)
|
||||
return True
|
||||
else:
|
||||
verbose_proxy_logger.info(
|
||||
"Spend tracking - pod %s could not acquire lock for cronjob_id=%s, "
|
||||
"held by pod %s. Spend updates in Redis will wait for the leader pod to commit.",
|
||||
self.pod_id,
|
||||
cronjob_id,
|
||||
current_value,
|
||||
)
|
||||
return False
|
||||
except Exception as e:
|
||||
verbose_proxy_logger.error(
|
||||
|
|
@ -124,10 +132,12 @@ class PodLockManager:
|
|||
pod_id=self.pod_id,
|
||||
)
|
||||
else:
|
||||
verbose_proxy_logger.debug(
|
||||
"Pod %s failed to release Redis lock for cronjob_id=%s",
|
||||
verbose_proxy_logger.warning(
|
||||
"Spend tracking - pod %s failed to release Redis lock for cronjob_id=%s. "
|
||||
"Lock will expire after TTL=%ds.",
|
||||
self.pod_id,
|
||||
cronjob_id,
|
||||
DEFAULT_CRON_JOB_LOCK_TTL_SECONDS,
|
||||
)
|
||||
else:
|
||||
verbose_proxy_logger.debug(
|
||||
|
|
|
|||
|
|
@ -96,14 +96,29 @@ class RedisUpdateBuffer:
|
|||
list_of_transactions = [safe_dumps(transactions)]
|
||||
if self.redis_cache is None:
|
||||
return
|
||||
current_redis_buffer_size = await self.redis_cache.async_rpush(
|
||||
key=redis_key,
|
||||
values=list_of_transactions,
|
||||
)
|
||||
await self._emit_new_item_added_to_redis_buffer_event(
|
||||
queue_size=current_redis_buffer_size,
|
||||
service=service_type,
|
||||
)
|
||||
try:
|
||||
current_redis_buffer_size = await self.redis_cache.async_rpush(
|
||||
key=redis_key,
|
||||
values=list_of_transactions,
|
||||
)
|
||||
verbose_proxy_logger.info(
|
||||
"Spend tracking - pushed spend updates to Redis buffer. "
|
||||
"redis_key=%s, buffer_size=%s",
|
||||
redis_key,
|
||||
current_redis_buffer_size,
|
||||
)
|
||||
await self._emit_new_item_added_to_redis_buffer_event(
|
||||
queue_size=current_redis_buffer_size,
|
||||
service=service_type,
|
||||
)
|
||||
except Exception as e:
|
||||
verbose_proxy_logger.error(
|
||||
"Spend tracking - failed to push spend updates to Redis (redis_key=%s). "
|
||||
"Error: %s",
|
||||
redis_key,
|
||||
str(e),
|
||||
)
|
||||
raise
|
||||
|
||||
async def store_in_memory_spend_updates_in_redis(
|
||||
self,
|
||||
|
|
@ -305,6 +320,13 @@ class RedisUpdateBuffer:
|
|||
if list_of_transactions is None:
|
||||
return None
|
||||
|
||||
verbose_proxy_logger.info(
|
||||
"Spend tracking - popped %d spend update batches from Redis buffer (key=%s). "
|
||||
"These items are now removed from Redis and must be committed to DB.",
|
||||
len(list_of_transactions) if isinstance(list_of_transactions, list) else 1,
|
||||
REDIS_UPDATE_BUFFER_KEY,
|
||||
)
|
||||
|
||||
# Parse the list of transactions from JSON strings
|
||||
parsed_transactions = self._parse_list_of_transactions(list_of_transactions)
|
||||
|
||||
|
|
|
|||
|
|
@ -31,6 +31,11 @@ class SpendUpdateQueue(BaseUpdateQueue):
|
|||
) -> DBSpendUpdateTransactions:
|
||||
"""Flush all updates from the queue and return all updates aggregated by entity type."""
|
||||
updates = await self.flush_all_updates_from_in_memory_queue()
|
||||
if len(updates) > 0:
|
||||
verbose_proxy_logger.info(
|
||||
"Spend tracking - flushed %d spend update items from in-memory queue",
|
||||
len(updates),
|
||||
)
|
||||
verbose_proxy_logger.debug("Aggregating updates by entity type: %s", updates)
|
||||
return self.get_aggregated_db_spend_update_transactions(updates)
|
||||
|
||||
|
|
|
|||
|
|
@ -1768,8 +1768,13 @@ async def update_cache( # noqa: PLR0915
|
|||
("{}:spend".format(litellm_proxy_admin_name), increment)
|
||||
)
|
||||
except Exception as e:
|
||||
verbose_proxy_logger.debug(
|
||||
f"An error occurred updating user cache: {str(e)}\n\n{traceback.format_exc()}"
|
||||
verbose_proxy_logger.warning(
|
||||
"Spend tracking - failed to update user spend in cache. "
|
||||
"Budget enforcement may use stale spend values. "
|
||||
"user_id=%s, response_cost=%s - %s",
|
||||
user_id,
|
||||
response_cost,
|
||||
str(e),
|
||||
)
|
||||
|
||||
### UPDATE END-USER SPEND ###
|
||||
|
|
@ -1806,8 +1811,13 @@ async def update_cache( # noqa: PLR0915
|
|||
existing_spend_obj.spend = new_spend
|
||||
values_to_update_in_cache.append((_id, existing_spend_obj.json()))
|
||||
except Exception as e:
|
||||
verbose_proxy_logger.exception(
|
||||
f"An error occurred updating end user cache: {str(e)}"
|
||||
verbose_proxy_logger.warning(
|
||||
"Spend tracking - failed to update end user spend in cache. "
|
||||
"Budget enforcement may use stale spend values. "
|
||||
"end_user_id=%s, response_cost=%s - %s",
|
||||
end_user_id,
|
||||
response_cost,
|
||||
str(e),
|
||||
)
|
||||
|
||||
### UPDATE TEAM SPEND ###
|
||||
|
|
@ -1848,8 +1858,13 @@ async def update_cache( # noqa: PLR0915
|
|||
existing_spend_obj.spend = new_spend
|
||||
values_to_update_in_cache.append((_id, existing_spend_obj))
|
||||
except Exception as e:
|
||||
verbose_proxy_logger.exception(
|
||||
f"An error occurred updating end user cache: {str(e)}"
|
||||
verbose_proxy_logger.warning(
|
||||
"Spend tracking - failed to update team spend in cache. "
|
||||
"Budget enforcement may use stale spend values. "
|
||||
"team_id=%s, response_cost=%s - %s",
|
||||
team_id,
|
||||
response_cost,
|
||||
str(e),
|
||||
)
|
||||
|
||||
### UPDATE TAG SPEND ###
|
||||
|
|
@ -1894,8 +1909,13 @@ async def update_cache( # noqa: PLR0915
|
|||
existing_tag_obj.spend = new_spend
|
||||
values_to_update_in_cache.append((cache_key, existing_tag_obj))
|
||||
except Exception as e:
|
||||
verbose_proxy_logger.exception(
|
||||
f"An error occurred updating tag cache: {str(e)}"
|
||||
verbose_proxy_logger.warning(
|
||||
"Spend tracking - failed to update tag spend in cache. "
|
||||
"Budget enforcement may use stale spend values. "
|
||||
"tags=%s, response_cost=%s - %s",
|
||||
tags,
|
||||
response_cost,
|
||||
str(e),
|
||||
)
|
||||
|
||||
if token is not None and response_cost is not None:
|
||||
|
|
|
|||
|
|
@ -4457,6 +4457,11 @@ class ProxyUpdateSpend:
|
|||
len(logs_to_process) :
|
||||
]
|
||||
popped_batch = True
|
||||
if len(logs_to_process) > 0:
|
||||
verbose_proxy_logger.info(
|
||||
"Spend tracking - processing %d spend logs for DB write",
|
||||
len(logs_to_process),
|
||||
)
|
||||
start_time = time.time()
|
||||
try:
|
||||
for i in range(n_retry_times + 1):
|
||||
|
|
@ -4503,9 +4508,17 @@ class ProxyUpdateSpend:
|
|||
f"{len(logs_to_process)} logs processed. Remaining in queue: {remaining_count}"
|
||||
)
|
||||
break
|
||||
except DB_CONNECTION_ERROR_TYPES:
|
||||
except DB_CONNECTION_ERROR_TYPES as e:
|
||||
if i is None:
|
||||
i = 0
|
||||
verbose_proxy_logger.warning(
|
||||
"Spend tracking - DB connection error writing spend logs, "
|
||||
"retry %d/%d. logs_count=%d, error=%s",
|
||||
i + 1,
|
||||
n_retry_times,
|
||||
len(logs_to_process),
|
||||
str(e),
|
||||
)
|
||||
if i >= n_retry_times:
|
||||
raise
|
||||
await asyncio.sleep(2**i)
|
||||
|
|
@ -4620,8 +4633,8 @@ async def update_spend_logs_job(
|
|||
logs_to_process=logs_to_process,
|
||||
)
|
||||
except Exception as guardrail_tracking_err:
|
||||
verbose_proxy_logger.debug(
|
||||
"Guardrail usage tracking failed (non-fatal): %s",
|
||||
verbose_proxy_logger.warning(
|
||||
"Spend tracking - guardrail usage tracking failed (non-fatal): %s",
|
||||
guardrail_tracking_err,
|
||||
)
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue