diff --git a/litellm/integrations/datadog/datadog_llm_obs.py b/litellm/integrations/datadog/datadog_llm_obs.py index adab0ca6593..7ab82eb7847 100644 --- a/litellm/integrations/datadog/datadog_llm_obs.py +++ b/litellm/integrations/datadog/datadog_llm_obs.py @@ -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]]: diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 2f21d280899..6ec3374c5ed 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -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 diff --git a/litellm/proxy/litellm_pre_call_utils.py b/litellm/proxy/litellm_pre_call_utils.py index 2ed3aed9bf9..4d885d92ad0 100644 --- a/litellm/proxy/litellm_pre_call_utils.py +++ b/litellm/proxy/litellm_pre_call_utils.py @@ -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 diff --git a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py index 6e5d4faabdb..f16bae559bb 100644 --- a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py @@ -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 diff --git a/litellm/types/integrations/datadog_llm_obs.py b/litellm/types/integrations/datadog_llm_obs.py index 25ee0932b12..1f281e93e8c 100644 --- a/litellm/types/integrations/datadog_llm_obs.py +++ b/litellm/types/integrations/datadog_llm_obs.py @@ -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 diff --git a/litellm/types/utils.py b/litellm/types/utils.py index 85e4e6bfdd4..1b633590a91 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -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] diff --git a/tests/test_litellm/integrations/datadog/test_datadog_llm_observability.py b/tests/test_litellm/integrations/datadog/test_datadog_llm_observability.py index 39b8427fcf1..e715fec4ffd 100644 --- a/tests/test_litellm/integrations/datadog/test_datadog_llm_observability.py +++ b/tests/test_litellm/integrations/datadog/test_datadog_llm_observability.py @@ -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