scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-06 02:48:13 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103

Merge pull request #14555 from mubashir1osmani/dd_spend_metric

Browse source 
DataDog shows spend metrics
This commit is contained in:
Krish Dholakia 2025-09-16 22:44:47 -07:00 • committed by GitHub
commit fcf84027e8
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 495 additions and 188 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

87
litellm/integrations/datadog/datadog_llm_obs.py
View file

@ -148,10 +148,22 @@ class DataDogLLMObsLogger(DataDogLogger, CustomBatchLogger):
),
),
}
verbose_logger.debug("payload %s", json.dumps(payload, indent=4))
# serialize datetime objects - for budget reset time in spend metrics
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
try:
verbose_logger.debug("payload %s", safe_dumps(payload))
except Exception as debug_error:
verbose_logger.debug(
"payload serialization failed: %s", str(debug_error)
)
json_payload = safe_dumps(payload)
response = await self.async_client.post(
url=self.intake_url,
json=payload,
content=json_payload,
headers={
"DD-API-KEY": self.DD_API_KEY,
"Content-Type": "application/json",
@ -494,6 +506,12 @@ class DataDogLLMObsLogger(DataDogLogger, CustomBatchLogger):
latency_metrics = self._get_latency_metrics(standard_logging_payload)
_metadata.update({"latency_metrics": dict(latency_metrics)})
#########################################################
# Add spend metrics to metadata
#########################################################
spend_metrics = self._get_spend_metrics(standard_logging_payload)
_metadata.update({"spend_metrics": dict(spend_metrics)})
## extract tool calls and add to metadata
tool_call_metadata = self._extract_tool_call_metadata(standard_logging_payload)
_metadata.update(tool_call_metadata)
@ -543,6 +561,71 @@ class DataDogLLMObsLogger(DataDogLogger, CustomBatchLogger):
return latency_metrics
def _get_spend_metrics(
self, standard_logging_payload: StandardLoggingPayload
) -> DDLLMObsSpendMetrics:
"""
Get the spend metrics from the standard logging payload
"""
spend_metrics: DDLLMObsSpendMetrics = DDLLMObsSpendMetrics()
# send response cost
spend_metrics["response_cost"] = standard_logging_payload.get(
"response_cost", 0.0
)
# Get budget information from metadata
metadata = standard_logging_payload.get("metadata", {})
# API key max budget
user_api_key_max_budget = metadata.get("user_api_key_max_budget")
if user_api_key_max_budget is not None:
spend_metrics["user_api_key_max_budget"] = float(user_api_key_max_budget)
# API key spend
user_api_key_spend = metadata.get("user_api_key_spend")
if user_api_key_spend is not None:
try:
spend_metrics["user_api_key_spend"] = float(user_api_key_spend)
except (ValueError, TypeError):
verbose_logger.debug(
f"Invalid user_api_key_spend value: {user_api_key_spend}"
)
# API key budget reset datetime
user_api_key_budget_reset_at = metadata.get("user_api_key_budget_reset_at")
if user_api_key_budget_reset_at is not None:
try:
from datetime import datetime, timezone
budget_reset_at = None
if isinstance(user_api_key_budget_reset_at, str):
# Handle ISO format strings that might have 'Z' suffix
iso_string = user_api_key_budget_reset_at.replace("Z", "+00:00")
budget_reset_at = datetime.fromisoformat(iso_string)
elif isinstance(user_api_key_budget_reset_at, datetime):
budget_reset_at = user_api_key_budget_reset_at
if budget_reset_at is not None:
# Preserve timezone info if already present
if budget_reset_at.tzinfo is None:
budget_reset_at = budget_reset_at.replace(tzinfo=timezone.utc)
# Convert to ISO string format for JSON serialization
# This prevents circular reference issues and ensures proper timezone representation
iso_string = budget_reset_at.isoformat()
spend_metrics["user_api_key_budget_reset_at"] = iso_string
# Debug logging to verify the conversion
verbose_logger.debug(
f"Converted budget_reset_at to ISO format: {iso_string}"
)
except Exception as e:
verbose_logger.debug(f"Error processing budget reset datetime: {e}")
verbose_logger.debug(f"Original value: {user_api_key_budget_reset_at}")
return spend_metrics
def _process_input_messages_preserving_tool_calls(
self, messages: List[Any]
) -> List[Dict[str, Any]]:

311
litellm/litellm_core_utils/litellm_logging.py
View file

@ -300,9 +300,9 @@ class Logging(LiteLLMLoggingBaseClass):
self.litellm_trace_id: str = litellm_trace_id or str(uuid.uuid4())
self.function_id = function_id
self.streaming_chunks: List[Any] = [] # for generating complete stream response
self.sync_streaming_chunks: List[Any] = (
[]
) # for generating complete stream response
self.sync_streaming_chunks: List[
Any
] = [] # for generating complete stream response
self.log_raw_request_response = log_raw_request_response
# Initialize dynamic callbacks
@ -672,24 +672,23 @@ class Logging(LiteLLMLoggingBaseClass):
if anthropic_cache_control_logger := AnthropicCacheControlHook.get_custom_logger_for_anthropic_cache_control_hook(
non_default_params
):
self.model_call_details["prompt_integration"] = (
anthropic_cache_control_logger.__class__.__name__
)
self.model_call_details[
"prompt_integration"
] = anthropic_cache_control_logger.__class__.__name__
return anthropic_cache_control_logger
#########################################################
# Vector Store / Knowledge Base hooks
#########################################################
if litellm.vector_store_registry is not None:
vector_store_custom_logger = _init_custom_logger_compatible_class(
logging_integration="vector_store_pre_call_hook",
internal_usage_cache=None,
llm_router=None,
)
self.model_call_details["prompt_integration"] = (
vector_store_custom_logger.__class__.__name__
)
self.model_call_details[
"prompt_integration"
] = vector_store_custom_logger.__class__.__name__
return vector_store_custom_logger
return None
@ -741,9 +740,9 @@ class Logging(LiteLLMLoggingBaseClass):
model
): # if model name was changes pre-call, overwrite the initial model call name with the new one
self.model_call_details["model"] = model
self.model_call_details["litellm_params"]["api_base"] = (
self._get_masked_api_base(additional_args.get("api_base", ""))
)
self.model_call_details["litellm_params"][
"api_base"
] = self._get_masked_api_base(additional_args.get("api_base", ""))
def pre_call(self, input, api_key, model=None, additional_args={}): # noqa: PLR0915
# Log the exact input to the LLM API
@ -772,10 +771,10 @@ class Logging(LiteLLMLoggingBaseClass):
try:
# [Non-blocking Extra Debug Information in metadata]
if turn_off_message_logging is True:
_metadata["raw_request"] = (
"redacted by litellm. \
_metadata[
"raw_request"
] = "redacted by litellm. \
'litellm.turn_off_message_logging=True'"
)
else:
curl_command = self._get_request_curl_command(
api_base=additional_args.get("api_base", ""),
@ -786,32 +785,32 @@ class Logging(LiteLLMLoggingBaseClass):
_metadata["raw_request"] = str(curl_command)
# split up, so it's easier to parse in the UI
self.model_call_details["raw_request_typed_dict"] = (
RawRequestTypedDict(
raw_request_api_base=str(
additional_args.get("api_base") or ""
),
raw_request_body=self._get_raw_request_body(
additional_args.get("complete_input_dict", {})
),
raw_request_headers=self._get_masked_headers(
additional_args.get("headers", {}) or {},
ignore_sensitive_headers=True,
),
error=None,
)
self.model_call_details[
"raw_request_typed_dict"
] = RawRequestTypedDict(
raw_request_api_base=str(
additional_args.get("api_base") or ""
),
raw_request_body=self._get_raw_request_body(
additional_args.get("complete_input_dict", {})
),
raw_request_headers=self._get_masked_headers(
additional_args.get("headers", {}) or {},
ignore_sensitive_headers=True,
),
error=None,
)
except Exception as e:
self.model_call_details["raw_request_typed_dict"] = (
RawRequestTypedDict(
error=str(e),
)
self.model_call_details[
"raw_request_typed_dict"
] = RawRequestTypedDict(
error=str(e),
)
_metadata["raw_request"] = (
"Unable to Log \
_metadata[
"raw_request"
] = "Unable to Log \
raw request: {}".format(
str(e)
)
str(e)
)
if getattr(self, "logger_fn", None) and callable(self.logger_fn):
try:
@ -1112,13 +1111,13 @@ class Logging(LiteLLMLoggingBaseClass):
for callback in callbacks:
try:
if isinstance(callback, CustomLogger):
response: Optional[MCPPostCallResponseObject] = (
await callback.async_post_mcp_tool_call_hook(
kwargs=kwargs,
response_obj=post_mcp_tool_call_response_obj,
start_time=start_time,
end_time=end_time,
)
response: Optional[
MCPPostCallResponseObject
] = await callback.async_post_mcp_tool_call_hook(
kwargs=kwargs,
response_obj=post_mcp_tool_call_response_obj,
start_time=start_time,
end_time=end_time,
)
######################################################################
# if any of the callbacks modify the response, use the modified response
@ -1238,9 +1237,9 @@ class Logging(LiteLLMLoggingBaseClass):
verbose_logger.debug(
f"response_cost_failure_debug_information: {debug_info}"
)
self.model_call_details["response_cost_failure_debug_information"] = (
debug_info
)
self.model_call_details[
"response_cost_failure_debug_information"
] = debug_info
return None
try:
@ -1265,9 +1264,9 @@ class Logging(LiteLLMLoggingBaseClass):
verbose_logger.debug(
f"response_cost_failure_debug_information: {debug_info}"
)
self.model_call_details["response_cost_failure_debug_information"] = (
debug_info
)
self.model_call_details[
"response_cost_failure_debug_information"
] = debug_info
return None
@ -1411,9 +1410,9 @@ class Logging(LiteLLMLoggingBaseClass):
end_time = datetime.datetime.now()
if self.completion_start_time is None:
self.completion_start_time = end_time
self.model_call_details["completion_start_time"] = (
self.completion_start_time
)
self.model_call_details[
"completion_start_time"
] = self.completion_start_time
self.model_call_details["log_event_type"] = "successful_api_call"
self.model_call_details["end_time"] = end_time
self.model_call_details["cache_hit"] = cache_hit
@ -1466,39 +1465,39 @@ class Logging(LiteLLMLoggingBaseClass):
"response_cost"
]
else:
self.model_call_details["response_cost"] = (
self._response_cost_calculator(result=logging_result)
)
self.model_call_details[
"response_cost"
] = self._response_cost_calculator(result=logging_result)
## STANDARDIZED LOGGING PAYLOAD
self.model_call_details["standard_logging_object"] = (
get_standard_logging_object_payload(
kwargs=self.model_call_details,
init_response_obj=logging_result,
start_time=start_time,
end_time=end_time,
logging_obj=self,
status="success",
standard_built_in_tools_params=self.standard_built_in_tools_params,
)
self.model_call_details[
"standard_logging_object"
] = get_standard_logging_object_payload(
kwargs=self.model_call_details,
init_response_obj=logging_result,
start_time=start_time,
end_time=end_time,
logging_obj=self,
status="success",
standard_built_in_tools_params=self.standard_built_in_tools_params,
)
elif isinstance(result, dict) or isinstance(result, list):
## STANDARDIZED LOGGING PAYLOAD
self.model_call_details["standard_logging_object"] = (
get_standard_logging_object_payload(
kwargs=self.model_call_details,
init_response_obj=result,
start_time=start_time,
end_time=end_time,
logging_obj=self,
status="success",
standard_built_in_tools_params=self.standard_built_in_tools_params,
)
self.model_call_details[
"standard_logging_object"
] = get_standard_logging_object_payload(
kwargs=self.model_call_details,
init_response_obj=result,
start_time=start_time,
end_time=end_time,
logging_obj=self,
status="success",
standard_built_in_tools_params=self.standard_built_in_tools_params,
)
elif standard_logging_object is not None:
self.model_call_details["standard_logging_object"] = (
standard_logging_object
)
self.model_call_details[
"standard_logging_object"
] = standard_logging_object
else: # streaming chunks + image gen.
self.model_call_details["response_cost"] = None
@ -1597,7 +1596,6 @@ class Logging(LiteLLMLoggingBaseClass):
)
if complete_streaming_response is not None:
self.success_handler(result=complete_streaming_response)
return
@ -1650,23 +1648,23 @@ class Logging(LiteLLMLoggingBaseClass):
verbose_logger.debug(
"Logging Details LiteLLM-Success Call streaming complete"
)
self.model_call_details["complete_streaming_response"] = (
complete_streaming_response
)
self.model_call_details["response_cost"] = (
self._response_cost_calculator(result=complete_streaming_response)
)
self.model_call_details[
"complete_streaming_response"
] = complete_streaming_response
self.model_call_details[
"response_cost"
] = self._response_cost_calculator(result=complete_streaming_response)
## STANDARDIZED LOGGING PAYLOAD
self.model_call_details["standard_logging_object"] = (
get_standard_logging_object_payload(
kwargs=self.model_call_details,
init_response_obj=complete_streaming_response,
start_time=start_time,
end_time=end_time,
logging_obj=self,
status="success",
standard_built_in_tools_params=self.standard_built_in_tools_params,
)
self.model_call_details[
"standard_logging_object"
] = get_standard_logging_object_payload(
kwargs=self.model_call_details,
init_response_obj=complete_streaming_response,
start_time=start_time,
end_time=end_time,
logging_obj=self,
status="success",
standard_built_in_tools_params=self.standard_built_in_tools_params,
)
callbacks = self.get_combined_callback_list(
dynamic_success_callbacks=self.dynamic_success_callbacks,
@ -1994,10 +1992,10 @@ class Logging(LiteLLMLoggingBaseClass):
)
else:
if self.stream and complete_streaming_response:
self.model_call_details["complete_response"] = (
self.model_call_details.get(
"complete_streaming_response", {}
)
self.model_call_details[
"complete_response"
] = self.model_call_details.get(
"complete_streaming_response", {}
)
result = self.model_call_details["complete_response"]
openMeterLogger.log_success_event(
@ -2036,10 +2034,10 @@ class Logging(LiteLLMLoggingBaseClass):
)
else:
if self.stream and complete_streaming_response:
self.model_call_details["complete_response"] = (
self.model_call_details.get(
"complete_streaming_response", {}
)
self.model_call_details[
"complete_response"
] = self.model_call_details.get(
"complete_streaming_response", {}
)
result = self.model_call_details["complete_response"]
@ -2141,10 +2139,12 @@ class Logging(LiteLLMLoggingBaseClass):
result.usage = batch_usage
elif not is_base64_unified_file_id: # only run for non-unified file ids
response_cost, batch_usage, batch_models = (
await _handle_completed_batch(
batch=result, custom_llm_provider=self.custom_llm_provider
)
(
response_cost,
batch_usage,
batch_models,
) = await _handle_completed_batch(
batch=result, custom_llm_provider=self.custom_llm_provider
)
result._hidden_params["response_cost"] = response_cost
@ -2175,9 +2175,9 @@ class Logging(LiteLLMLoggingBaseClass):
if complete_streaming_response is not None:
print_verbose("Async success callbacks: Got a complete streaming response")
self.model_call_details["async_complete_streaming_response"] = (
complete_streaming_response
)
self.model_call_details[
"async_complete_streaming_response"
] = complete_streaming_response
try:
if self.model_call_details.get("cache_hit", False) is True:
@ -2188,10 +2188,10 @@ class Logging(LiteLLMLoggingBaseClass):
model_call_details=self.model_call_details
)
# base_model defaults to None if not set on model_info
self.model_call_details["response_cost"] = (
self._response_cost_calculator(
result=complete_streaming_response
)
self.model_call_details[
"response_cost"
] = self._response_cost_calculator(
result=complete_streaming_response
)
verbose_logger.debug(
@ -2204,16 +2204,16 @@ class Logging(LiteLLMLoggingBaseClass):
self.model_call_details["response_cost"] = None
## STANDARDIZED LOGGING PAYLOAD
self.model_call_details["standard_logging_object"] = (
get_standard_logging_object_payload(
kwargs=self.model_call_details,
init_response_obj=complete_streaming_response,
start_time=start_time,
end_time=end_time,
logging_obj=self,
status="success",
standard_built_in_tools_params=self.standard_built_in_tools_params,
)
self.model_call_details[
"standard_logging_object"
] = get_standard_logging_object_payload(
kwargs=self.model_call_details,
init_response_obj=complete_streaming_response,
start_time=start_time,
end_time=end_time,
logging_obj=self,
status="success",
standard_built_in_tools_params=self.standard_built_in_tools_params,
)
callbacks = self.get_combined_callback_list(
dynamic_success_callbacks=self.dynamic_async_success_callbacks,
@ -2426,18 +2426,18 @@ class Logging(LiteLLMLoggingBaseClass):
## STANDARDIZED LOGGING PAYLOAD
self.model_call_details["standard_logging_object"] = (
get_standard_logging_object_payload(
kwargs=self.model_call_details,
init_response_obj={},
start_time=start_time,
end_time=end_time,
logging_obj=self,
status="failure",
error_str=str(exception),
original_exception=exception,
standard_built_in_tools_params=self.standard_built_in_tools_params,
)
self.model_call_details[
"standard_logging_object"
] = get_standard_logging_object_payload(
kwargs=self.model_call_details,
init_response_obj={},
start_time=start_time,
end_time=end_time,
logging_obj=self,
status="failure",
error_str=str(exception),
original_exception=exception,
standard_built_in_tools_params=self.standard_built_in_tools_params,
)
return start_time, end_time
@ -3326,9 +3326,9 @@ def _init_custom_logger_compatible_class( # noqa: PLR0915
endpoint=arize_config.endpoint,
)
os.environ["OTEL_EXPORTER_OTLP_TRACES_HEADERS"] = (
f"space_id={arize_config.space_key},api_key={arize_config.api_key}"
)
os.environ[
"OTEL_EXPORTER_OTLP_TRACES_HEADERS"
] = f"space_id={arize_config.space_key},api_key={arize_config.api_key}"
for callback in _in_memory_loggers:
if (
isinstance(callback, ArizeLogger)
@ -3352,9 +3352,9 @@ def _init_custom_logger_compatible_class( # noqa: PLR0915
# auth can be disabled on local deployments of arize phoenix
if arize_phoenix_config.otlp_auth_headers is not None:
os.environ["OTEL_EXPORTER_OTLP_TRACES_HEADERS"] = (
arize_phoenix_config.otlp_auth_headers
)
os.environ[
"OTEL_EXPORTER_OTLP_TRACES_HEADERS"
] = arize_phoenix_config.otlp_auth_headers
for callback in _in_memory_loggers:
if (
@ -3462,9 +3462,9 @@ def _init_custom_logger_compatible_class( # noqa: PLR0915
exporter="otlp_http",
endpoint="https://langtrace.ai/api/trace",
)
os.environ["OTEL_EXPORTER_OTLP_TRACES_HEADERS"] = (
f"api_key={os.getenv('LANGTRACE_API_KEY')}"
)
os.environ[
"OTEL_EXPORTER_OTLP_TRACES_HEADERS"
] = f"api_key={os.getenv('LANGTRACE_API_KEY')}"
for callback in _in_memory_loggers:
if (
isinstance(callback, OpenTelemetry)
@ -4114,10 +4114,10 @@ class StandardLoggingPayloadSetup:
for key in StandardLoggingHiddenParams.__annotations__.keys():
if key in hidden_params:
if key == "additional_headers":
clean_hidden_params["additional_headers"] = (
StandardLoggingPayloadSetup.get_additional_headers(
hidden_params[key]
)
clean_hidden_params[
"additional_headers"
] = StandardLoggingPayloadSetup.get_additional_headers(
hidden_params[key]
)
else:
clean_hidden_params[key] = hidden_params[key] # type: ignore
@ -4553,6 +4553,9 @@ def get_standard_logging_metadata(
clean_metadata = StandardLoggingMetadata(
user_api_key_hash=None,
user_api_key_alias=None,
user_api_key_spend=None,
user_api_key_max_budget=None,
user_api_key_budget_reset_at=None,
user_api_key_team_id=None,
user_api_key_org_id=None,
user_api_key_user_id=None,
@ -4602,9 +4605,9 @@ def scrub_sensitive_keys_in_metadata(litellm_params: Optional[dict]):
):
for k, v in metadata["user_api_key_metadata"].items():
if k == "logging": # prevent logging user logging keys
cleaned_user_api_key_metadata[k] = (
"scrubbed_by_litellm_for_sensitive_keys"
)
cleaned_user_api_key_metadata[
k
] = "scrubbed_by_litellm_for_sensitive_keys"
else:
cleaned_user_api_key_metadata[k] = v

31
litellm/proxy/litellm_pre_call_utils.py
View file

@ -169,12 +169,12 @@ def _get_dynamic_logging_metadata(
user_api_key_dict: UserAPIKeyAuth, proxy_config: ProxyConfig
) -> Optional[TeamCallbackMetadata]:
callback_settings_obj: Optional[TeamCallbackMetadata] = None
key_dynamic_logging_settings: Optional[dict] = (
KeyAndTeamLoggingSettings.get_key_dynamic_logging_settings(user_api_key_dict)
)
team_dynamic_logging_settings: Optional[dict] = (
KeyAndTeamLoggingSettings.get_team_dynamic_logging_settings(user_api_key_dict)
)
key_dynamic_logging_settings: Optional[
dict
] = KeyAndTeamLoggingSettings.get_key_dynamic_logging_settings(user_api_key_dict)
team_dynamic_logging_settings: Optional[
dict
] = KeyAndTeamLoggingSettings.get_team_dynamic_logging_settings(user_api_key_dict)
#########################################################################################
# Key-based callbacks
#########################################################################################
@ -562,6 +562,8 @@ class LiteLLMProxyRequestSetup:
user_api_key_logged_metadata = StandardLoggingUserAPIKeyMetadata(
user_api_key_hash=user_api_key_dict.api_key, # just the hashed token
user_api_key_alias=user_api_key_dict.key_alias,
user_api_key_spend=user_api_key_dict.spend,
user_api_key_max_budget=user_api_key_dict.max_budget,
user_api_key_team_id=user_api_key_dict.team_id,
user_api_key_user_id=user_api_key_dict.user_id,
user_api_key_org_id=user_api_key_dict.org_id,
@ -569,6 +571,7 @@ class LiteLLMProxyRequestSetup:
user_api_key_end_user_id=user_api_key_dict.end_user_id,
user_api_key_user_email=user_api_key_dict.user_email,
user_api_key_request_route=user_api_key_dict.request_route,
user_api_key_budget_reset_at=user_api_key_dict.budget_reset_at,
)
return user_api_key_logged_metadata
@ -611,11 +614,11 @@ class LiteLLMProxyRequestSetup:
## KEY-LEVEL SPEND LOGS / TAGS
if "tags" in key_metadata and key_metadata["tags"] is not None:
data[_metadata_variable_name]["tags"] = (
LiteLLMProxyRequestSetup._merge_tags(
request_tags=data[_metadata_variable_name].get("tags"),
tags_to_add=key_metadata["tags"],
)
data[_metadata_variable_name][
"tags"
] = LiteLLMProxyRequestSetup._merge_tags(
request_tags=data[_metadata_variable_name].get("tags"),
tags_to_add=key_metadata["tags"],
)
if "spend_logs_metadata" in key_metadata and isinstance(
key_metadata["spend_logs_metadata"], dict
@ -844,9 +847,9 @@ async def add_litellm_data_to_request( # noqa: PLR0915
data[_metadata_variable_name]["litellm_api_version"] = version
if general_settings is not None:
data[_metadata_variable_name]["global_max_parallel_requests"] = (
general_settings.get("global_max_parallel_requests", None)
)
data[_metadata_variable_name][
"global_max_parallel_requests"
] = general_settings.get("global_max_parallel_requests", None)
### KEY-LEVEL Controls
key_metadata = user_api_key_dict.metadata

24
litellm/proxy/pass_through_endpoints/pass_through_endpoints.py
View file

@ -474,6 +474,9 @@ class HttpPassThroughEndpointHelpers(BasePassthroughUtils):
user_api_key_team_alias=user_api_key_dict.team_alias,
user_api_key_end_user_id=user_api_key_dict.end_user_id,
user_api_key_request_route=user_api_key_dict.request_route,
user_api_key_spend=user_api_key_dict.spend,
user_api_key_max_budget=user_api_key_dict.max_budget,
user_api_key_budget_reset_at=user_api_key_dict.budget_reset_at,
)
)
@ -1003,7 +1006,7 @@ class InitPassThroughEndpointHelpers:
):
"""Add exact path route for pass-through endpoint"""
route_key = f"{endpoint_id}:exact:{path}"
# Check if this exact route is already registered
if route_key in _registered_pass_through_routes:
verbose_proxy_logger.debug(
@ -1011,7 +1014,7 @@ class InitPassThroughEndpointHelpers:
path,
)
return
verbose_proxy_logger.debug(
"adding exact pass through endpoint: %s, dependencies: %s",
path,
@ -1032,12 +1035,12 @@ class InitPassThroughEndpointHelpers:
methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
dependencies=dependencies,
)
# Register the route to prevent duplicates
_registered_pass_through_routes[route_key] = {
"endpoint_id": endpoint_id,
"path": path,
"type": "exact"
"type": "exact",
}
@staticmethod
@ -1055,7 +1058,7 @@ class InitPassThroughEndpointHelpers:
"""Add wildcard route for sub-paths"""
wildcard_path = f"{path}/{{subpath:path}}"
route_key = f"{endpoint_id}:subpath:{path}"
# Check if this subpath route is already registered
if route_key in _registered_pass_through_routes:
verbose_proxy_logger.debug(
@ -1063,7 +1066,7 @@ class InitPassThroughEndpointHelpers:
wildcard_path,
)
return
verbose_proxy_logger.debug(
"adding wildcard pass through endpoint: %s, dependencies: %s",
wildcard_path,
@ -1085,19 +1088,20 @@ class InitPassThroughEndpointHelpers:
methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
dependencies=dependencies,
)
# Register the route to prevent duplicates
_registered_pass_through_routes[route_key] = {
"endpoint_id": endpoint_id,
"path": path,
"type": "subpath"
"type": "subpath",
}
@staticmethod
def remove_endpoint_routes(endpoint_id: str):
"""Remove all routes for a specific endpoint ID from the registry"""
keys_to_remove = [
key for key, value in _registered_pass_through_routes.items()
key
for key, value in _registered_pass_through_routes.items()
if value["endpoint_id"] == endpoint_id
]
for key in keys_to_remove:
@ -1480,7 +1484,7 @@ async def delete_pass_through_endpoints(
pass_through_endpoint_data.pop(endpoint_index)
response_obj = found_endpoint
# Remove routes from registry
# Remove routes from registry
InitPassThroughEndpointHelpers.remove_endpoint_routes(endpoint_id)
## Update db

7
litellm/types/integrations/datadog_llm_obs.py
View file

@ -83,3 +83,10 @@ class DDLLMObsLatencyMetrics(TypedDict, total=False):
time_to_first_token_ms: float
litellm_overhead_time_ms: float
guardrail_overhead_time_ms: float
class DDLLMObsSpendMetrics(TypedDict, total=False):
response_cost: float
user_api_key_spend: float
user_api_key_max_budget: float
user_api_key_budget_reset_at: str

3
litellm/types/utils.py
View file

@ -1810,6 +1810,9 @@ class AdapterCompletionStreamWrapper:
class StandardLoggingUserAPIKeyMetadata(TypedDict):
user_api_key_hash: Optional[str] # hash of the litellm virtual key used
user_api_key_alias: Optional[str]
user_api_key_spend: Optional[float]
user_api_key_max_budget: Optional[float]
user_api_key_budget_reset_at: Optional[str]
user_api_key_org_id: Optional[str]
user_api_key_team_id: Optional[str]
user_api_key_user_id: Optional[str]

220
tests/test_litellm/integrations/datadog/test_datadog_llm_observability.py
View file

@ -1,7 +1,7 @@
import asyncio
import os
import sys
from datetime import datetime, timedelta
from datetime import datetime, timedelta, timezone
from typing import Optional
from unittest.mock import Mock, patch, MagicMock
@ -661,17 +661,24 @@ def create_standard_logging_payload_with_tool_calls() -> StandardLoggingPayload:
"""Create a StandardLoggingPayload object with tool calls for testing"""
return {
"id": "test-request-id-tool-calls",
"trace_id": "test-trace-id-tool-calls",
"call_type": "completion",
"stream": None,
"response_cost": 0.05,
"response_cost_failure_debug_info": None,
"status": "success",
"custom_llm_provider": "openai",
"total_tokens": 50,
"prompt_tokens": 20,
"completion_tokens": 30,
"startTime": 1234567890.0,
"endTime": 1234567891.0,
"completionStartTime": 1234567890.5,
"model_map_information": {"model_map_key": "gpt-4", "model_map_value": None},
"response_time": 1.0,
"model_map_information": {
"model_map_key": "gpt-4",
"model_map_value": None
},
"model": "gpt-4",
"model_id": "model-123",
"model_group": "openai-gpt",
@ -746,6 +753,7 @@ def create_standard_logging_payload_with_tool_calls() -> StandardLoggingPayload:
]
},
"error_str": None,
"error_information": None,
"model_parameters": {"temperature": 0.7},
"hidden_params": {
"model_id": "model-123",
@ -758,14 +766,9 @@ def create_standard_logging_payload_with_tool_calls() -> StandardLoggingPayload:
"litellm_model_name": None,
"usage_object": None,
},
"stream": None,
"response_time": 1.0,
"error_information": None,
"guardrail_information": None,
"standard_built_in_tools_params": None,
"trace_id": "test-trace-id-tool-calls",
"custom_llm_provider": "openai",
}
} # type: ignore
class TestDataDogLLMObsLoggerToolCalls:
@ -897,3 +900,204 @@ class TestDataDogLLMObsLoggerToolCalls:
assert len(output_tool_calls) == 1
output_function_info = output_tool_calls[0].get("function", {})
assert output_function_info.get("name") == "format_response"
def create_standard_logging_payload_with_spend_metrics() -> StandardLoggingPayload:
"""Create a StandardLoggingPayload object with spend metrics for testing"""
from datetime import datetime, timezone
# Create a budget reset time 10 days from now (using "10d" format)
budget_reset_at = datetime.now(timezone.utc) + timedelta(days=10)
return {
"id": "test-request-id-spend",
"trace_id": "test-trace-id-spend",
"call_type": "completion",
"stream": None,
"response_cost": 0.15,
"response_cost_failure_debug_info": None,
"status": "success",
"custom_llm_provider": "openai",
"total_tokens": 30,
"prompt_tokens": 10,
"completion_tokens": 20,
"startTime": 1234567890.0,
"endTime": 1234567891.0,
"completionStartTime": 1234567890.5,
"response_time": 1.0,
"model_map_information": {
"model_map_key": "gpt-4",
"model_map_value": None
},
"model": "gpt-4",
"model_id": "model-123",
"model_group": "openai-gpt",
"api_base": "https://api.openai.com",
"metadata": {
"user_api_key_hash": "test_hash",
"user_api_key_org_id": None,
"user_api_key_alias": "test_alias",
"user_api_key_team_id": "test_team",
"user_api_key_user_id": "test_user",
"user_api_key_team_alias": "test_team_alias",
"user_api_key_user_email": None,
"user_api_key_end_user_id": None,
"user_api_key_request_route": None,
"user_api_key_spend": 0.67,
"user_api_key_max_budget": 10.0, # $10 max budget
"user_api_key_budget_reset_at": budget_reset_at.isoformat(), # ISO format: 2025-09-26T...
"spend_logs_metadata": None,
"requester_ip_address": "127.0.0.1",
"requester_metadata": None,
"requester_custom_headers": None,
"prompt_management_metadata": None,
"mcp_tool_call_metadata": None,
"vector_store_request_metadata": None,
"applied_guardrails": None,
"usage_object": None,
"cold_storage_object_key": None,
},
"cache_hit": False,
"cache_key": None,
"saved_cache_cost": 0.0,
"request_tags": [],
"end_user": None,
"requester_ip_address": "127.0.0.1",
"messages": [{"role": "user", "content": "Hello, world!"}],
"response": {"choices": [{"message": {"content": "Hi there!"}}]},
"error_str": None,
"error_information": None,
"model_parameters": {"stream": False},
"hidden_params": {
"model_id": "model-123",
"cache_key": None,
"api_base": "https://api.openai.com",
"response_cost": "0.15",
"litellm_overhead_time_ms": None,
"additional_headers": None,
"batch_models": None,
"litellm_model_name": None,
"usage_object": None,
},
"guardrail_information": None,
"standard_built_in_tools_params": None,
} # type: ignore
@pytest.mark.asyncio
async def test_datadog_llm_obs_spend_metrics(mock_env_vars):
"""Test that budget metrics are properly extracted and logged"""
datadog_llm_obs_logger = DataDogLLMObsLogger()
# Create a standard logging payload with spend metrics
payload = create_standard_logging_payload_with_spend_metrics()
# Show the budget reset time in ISO format
budget_reset_iso = payload["metadata"]["user_api_key_budget_reset_at"]
print(f"Budget reset time (ISO format): {budget_reset_iso}")
from datetime import datetime, timezone
print(f"Current time: {datetime.now(timezone.utc).isoformat()}")
# Test the _get_spend_metrics method
spend_metrics = datadog_llm_obs_logger._get_spend_metrics(payload)
# Verify budget metrics are present
assert "user_api_key_max_budget" in spend_metrics
assert spend_metrics["user_api_key_max_budget"] == 10.0
assert "user_api_key_budget_reset_at" in spend_metrics
# The budget reset should be a datetime string in ISO format
budget_reset = spend_metrics["user_api_key_budget_reset_at"]
assert isinstance(budget_reset, str)
print(f"Budget reset datetime: {budget_reset}")
# Should be close to 10 days from now
budget_reset_dt = datetime.fromisoformat(budget_reset.replace('Z', '+00:00'))
now = datetime.now(timezone.utc)
time_diff = (budget_reset_dt - now).total_seconds() / 86400 # days
assert 9.5 <= time_diff <= 10.5 # Should be close to 10 days
print(f"Spend metrics: {spend_metrics}")
@pytest.mark.asyncio
async def test_datadog_llm_obs_spend_metrics_no_budget(mock_env_vars):
"""Test that spend metrics work when no budget is set"""
datadog_llm_obs_logger = DataDogLLMObsLogger()
# Create a standard logging payload without budget metadata
payload = create_standard_logging_payload_with_spend_metrics()
# Remove budget-related metadata to test no-budget scenario
payload["metadata"].pop("user_api_key_max_budget", None)
payload["metadata"].pop("user_api_key_budget_reset_at", None)
# Test the _get_spend_metrics method
spend_metrics = datadog_llm_obs_logger._get_spend_metrics(payload)
# Verify only response cost is present
assert "response_cost" in spend_metrics
assert spend_metrics["response_cost"] == 0.15
# Budget metrics should not be present
assert "user_api_key_max_budget" not in spend_metrics
assert "user_api_key_budget_reset_at" not in spend_metrics
print(f"Spend metrics (no budget): {spend_metrics}")
@pytest.mark.asyncio
async def test_spend_metrics_in_datadog_payload(mock_env_vars):
"""Test that spend metrics are correctly included in DataDog LLM Observability payloads"""
from datetime import datetime
datadog_llm_obs_logger = DataDogLLMObsLogger()
standard_payload = create_standard_logging_payload_with_spend_metrics()
kwargs = {
"standard_logging_object": standard_payload,
"litellm_params": {"metadata": {}},
}
start_time = datetime.now()
end_time = datetime.now()
payload = datadog_llm_obs_logger.create_llm_obs_payload(kwargs, start_time, end_time)
# Verify basic payload structure
assert payload.get("name") == "litellm_llm_call"
assert payload.get("status") == "ok"
# Verify spend metrics are included in metadata
meta = payload.get("meta", {})
assert meta is not None, "Meta section should exist in payload"
metadata = meta.get("metadata", {})
assert metadata is not None, "Metadata section should exist in meta"
spend_metrics = metadata.get("spend_metrics", {})
assert spend_metrics, "Spend metrics should exist in metadata"
# Check that all metrics are present
assert "response_cost" in spend_metrics
assert "user_api_key_spend" in spend_metrics
assert "user_api_key_max_budget" in spend_metrics
assert "user_api_key_budget_reset_at" in spend_metrics
# Verify the values are correct
assert spend_metrics["response_cost"] == 0.15 # response_cost
assert spend_metrics["user_api_key_spend"] == 0.67 # lol
assert spend_metrics["user_api_key_max_budget"] == 10.0 # max budget
# Verify budget reset is a datetime string in ISO format
budget_reset = spend_metrics["user_api_key_budget_reset_at"]
assert isinstance(budget_reset, str)
print(f"Budget reset in payload: {budget_reset}") # In StandardLoggingUserAPIKeyMetadata
user_api_key_budget_reset_at: Optional[str] = None
# In DDLLMObsSpendMetrics
user_api_key_budget_reset_at: str
# Should be close to 10 days from now
from datetime import datetime, timezone
budget_reset_dt = datetime.fromisoformat(budget_reset.replace('Z', '+00:00'))
now = datetime.now(timezone.utc)
time_diff = (budget_reset_dt - now).total_seconds() / 86400 # days
assert 9.5 <= time_diff <= 10.5 # Should be close to 10 days