From ef08e18c6641cebc97cb2d9b419d8bb257b25c5c Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Mon, 18 Aug 2025 15:19:48 -0700 Subject: [PATCH] [Feat] Datadog LLM Observability - Add support for Failure Logging (#13726) * add async_log_failure_event for DD LLM Obs * update types * DataDogLLMObsLogger add failure logging support * test_async_log_failure_event * dd test failure --- .../integrations/datadog/datadog_llm_obs.py | 61 +++++++++- litellm/proxy/proxy_config.yaml | 2 + litellm/types/integrations/datadog_llm_obs.py | 17 ++- .../datadog/test_datadog_llm_observability.py | 115 +++++++++++++++++- 4 files changed, 186 insertions(+), 9 deletions(-) diff --git a/litellm/integrations/datadog/datadog_llm_obs.py b/litellm/integrations/datadog/datadog_llm_obs.py index 2577ed3ddf..a33fb2dfee 100644 --- a/litellm/integrations/datadog/datadog_llm_obs.py +++ b/litellm/integrations/datadog/datadog_llm_obs.py @@ -27,7 +27,11 @@ from litellm.llms.custom_httpx.http_handler import ( httpxSpecialProvider, ) from litellm.types.integrations.datadog_llm_obs import * -from litellm.types.utils import CallTypes, StandardLoggingPayload +from litellm.types.utils import ( + CallTypes, + StandardLoggingPayload, + StandardLoggingPayloadErrorInformation, +) class DataDogLLMObsLogger(DataDogLogger, CustomBatchLogger): @@ -102,6 +106,24 @@ class DataDogLLMObsLogger(DataDogLogger, CustomBatchLogger): verbose_logger.exception( f"DataDogLLMObs: Error logging success event - {str(e)}" ) + + async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time): + try: + verbose_logger.debug( + f"DataDogLLMObs: Logging failure event for model {kwargs.get('model', 'unknown')}" + ) + payload = self.create_llm_obs_payload( + kwargs, start_time, end_time + ) + verbose_logger.debug(f"DataDogLLMObs: Payload: {payload}") + self.log_queue.append(payload) + + if len(self.log_queue) >= self.batch_size: + await self.async_send_batch() + except Exception as e: + verbose_logger.exception( + f"DataDogLLMObs: Error logging failure event - {str(e)}" + ) async def async_send_batch(self): try: @@ -174,11 +196,14 @@ class DataDogLLMObsLogger(DataDogLogger, CustomBatchLogger): call_type=standard_logging_payload.get("call_type") )) + error_info = self._assemble_error_info(standard_logging_payload) + meta = Meta( kind=self._get_datadog_span_kind(standard_logging_payload.get("call_type")), input=input_meta, output=output_meta, metadata=self._get_dd_llm_obs_payload_metadata(standard_logging_payload), + error=error_info, ) # Calculate metrics (you may need to adjust these based on available data) @@ -199,11 +224,31 @@ class DataDogLLMObsLogger(DataDogLogger, CustomBatchLogger): start_ns=int(start_time.timestamp() * 1e9), duration=int((end_time - start_time).total_seconds() * 1e9), metrics=metrics, + status="error" if error_info else "ok", tags=[ self._get_datadog_tags(standard_logging_object=standard_logging_payload) ], ) + def _assemble_error_info(self, standard_logging_payload: StandardLoggingPayload) -> Optional[DDLLMObsError]: + """ + Assemble error information for failure cases according to DD LLM Obs API spec + """ + # Handle error information for failure cases according to DD LLM Obs API spec + error_info: Optional[DDLLMObsError] = None + + if standard_logging_payload.get("status") == "failure": + # Try to get structured error information first + error_information: Optional[StandardLoggingPayloadErrorInformation] = standard_logging_payload.get("error_information") + + if error_information: + error_info = DDLLMObsError( + message=error_information.get("error_message") or standard_logging_payload.get("error_str") or "Unknown error", + type=error_information.get("error_class"), + stack=error_information.get("traceback") + ) + return error_info + def _get_time_to_first_token_seconds(self, standard_logging_payload: StandardLoggingPayload) -> float: """ Get the time to first token in seconds @@ -232,8 +277,20 @@ class DataDogLLMObsLogger(DataDogLogger, CustomBatchLogger): for now this handles logging /chat/completions responses """ + if response_obj is None: + return [] + if call_type in [CallTypes.completion.value, CallTypes.acompletion.value]: - return [response_obj["choices"][0]["message"]] + try: + # Safely extract message from response_obj, handle failure cases + if isinstance(response_obj, dict) and "choices" in response_obj: + choices = response_obj["choices"] + if choices and len(choices) > 0 and "message" in choices[0]: + return [choices[0]["message"]] + return [] + except (KeyError, IndexError, TypeError): + # In case of any error accessing the response structure, return empty list + return [] return [] def _get_datadog_span_kind(self, call_type: Optional[str]) -> Literal["llm", "tool", "task", "embedding", "retrieval"]: diff --git a/litellm/proxy/proxy_config.yaml b/litellm/proxy/proxy_config.yaml index 5e1e3ea28e..ef15e3f3ba 100644 --- a/litellm/proxy/proxy_config.yaml +++ b/litellm/proxy/proxy_config.yaml @@ -5,3 +5,5 @@ model_list: - model_name: bedrock/us.anthropic.claude-sonnet-4-20250514-v1:0 litellm_params: model: bedrock/us.anthropic.claude-sonnet-4-20250514-v1:0 +litellm_settings: + callbacks: ["datadog_llm_observability"] \ No newline at end of file diff --git a/litellm/types/integrations/datadog_llm_obs.py b/litellm/types/integrations/datadog_llm_obs.py index 25685db483..008101cec1 100644 --- a/litellm/types/integrations/datadog_llm_obs.py +++ b/litellm/types/integrations/datadog_llm_obs.py @@ -18,12 +18,20 @@ class OutputMeta(TypedDict): messages: List[Any] -class Meta(TypedDict): +class DDLLMObsError(TypedDict, total=False): + """Error information on the span according to DD LLM Obs API spec""" + message: str # The error message + stack: Optional[str] # The stack trace + type: Optional[str] # The error type + + +class Meta(TypedDict, total=False): # The span kind: "agent", "workflow", "llm", "tool", "task", "embedding", or "retrieval". kind: Literal["llm", "tool", "task", "embedding", "retrieval"] - input: InputMeta # The span’s input information. - output: OutputMeta # The span’s output information. + input: InputMeta # The span's input information. + output: OutputMeta # The span's output information. metadata: Dict[str, Any] + error: Optional[DDLLMObsError] # Error information on the span class LLMMetrics(TypedDict, total=False): @@ -35,7 +43,7 @@ class LLMMetrics(TypedDict, total=False): total_cost: float -class LLMObsPayload(TypedDict): +class LLMObsPayload(TypedDict, total=False): parent_id: str trace_id: str span_id: str @@ -45,6 +53,7 @@ class LLMObsPayload(TypedDict): duration: int metrics: LLMMetrics tags: List + status: Literal["ok", "error"] # Error status ("ok" or "error"). Defaults to "ok". class DDSpanAttributes(TypedDict): 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 18d3efdddd..c53ea270f4 100644 --- a/tests/test_litellm/integrations/datadog/test_datadog_llm_observability.py +++ b/tests/test_litellm/integrations/datadog/test_datadog_llm_observability.py @@ -24,6 +24,7 @@ from litellm.types.utils import ( StandardLoggingMetadata, StandardLoggingModelInformation, StandardLoggingPayload, + StandardLoggingPayloadErrorInformation, ) @@ -81,6 +82,67 @@ def create_standard_logging_payload_with_cache() -> StandardLoggingPayload: ) +def create_standard_logging_payload_with_failure() -> StandardLoggingPayload: + """Create a StandardLoggingPayload object for failure testing""" + return StandardLoggingPayload( + id="test-request-id-failure-789", + call_type="completion", + response_cost=0.0, + response_cost_failure_debug_info=None, + status="failure", + total_tokens=0, + prompt_tokens=10, + completion_tokens=0, + startTime=1234567890.0, + endTime=1234567891.0, + completionStartTime=1234567890.5, + model_map_information=StandardLoggingModelInformation( + 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=StandardLoggingMetadata( + 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", + spend_logs_metadata=None, + requester_ip_address="127.0.0.1", + requester_metadata=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=None, + error_str="RateLimitError: You exceeded your current quota", + error_information=StandardLoggingPayloadErrorInformation( + error_code="rate_limit_exceeded", + error_class="RateLimitError", + llm_provider="openai", + traceback="Traceback (most recent call last):\n File test.py, line 1\n RateLimitError: You exceeded your current quota", + error_message="RateLimitError: You exceeded your current quota" + ), + model_parameters={"stream": False}, + hidden_params=StandardLoggingHiddenParams( + model_id="model-123", + cache_key=None, + api_base="https://api.openai.com", + response_cost="0.0", + additional_headers=None, + ), + trace_id="test-trace-id-failure-456", + custom_llm_provider="openai", + ) + + class TestDataDogLLMObsLogger: """Test suite for DataDog LLM Observability Logger""" @@ -118,7 +180,7 @@ class TestDataDogLLMObsLogger: start_time = datetime.now() end_time = datetime.now() - payload = logger.create_llm_obs_payload(kwargs, mock_response_obj, start_time, end_time) + payload = logger.create_llm_obs_payload(kwargs, start_time, end_time) # Test 1: Verify total_cost is correctly extracted from response_cost assert payload["metrics"].get("total_cost") == 0.05 @@ -148,7 +210,7 @@ class TestDataDogLLMObsLogger: start_time = datetime.now() end_time = datetime.now() - payload = logger.create_llm_obs_payload(kwargs, mock_response_obj, start_time, end_time) + payload = logger.create_llm_obs_payload(kwargs, start_time, end_time) # Test the _get_dd_llm_obs_payload_metadata method directly metadata = logger._get_dd_llm_obs_payload_metadata(standard_payload) @@ -217,9 +279,56 @@ class TestDataDogLLMObsLogger: assert logger._get_datadog_span_kind("unknown_call_type") == "llm" assert logger._get_datadog_span_kind(None) == "llm" + @pytest.mark.asyncio + async def test_async_log_failure_event(self, mock_env_vars): + """Test that async_log_failure_event correctly processes failure payloads according to DD LLM Obs API spec""" + with patch('litellm.integrations.datadog.datadog_llm_obs.get_async_httpx_client'), \ + patch('asyncio.create_task'): + logger = DataDogLLMObsLogger() + + # Ensure log_queue starts empty + logger.log_queue = [] + + standard_failure_payload = create_standard_logging_payload_with_failure() + + kwargs = { + "standard_logging_object": standard_failure_payload, + "model": "gpt-4", + "litellm_params": {"metadata": {}} + } + + start_time = datetime.now() + end_time = datetime.now() + timedelta(seconds=2) + + # Mock async_send_batch to prevent actual network calls + with patch.object(logger, 'async_send_batch') as mock_send_batch: + # Call the method under test + await logger.async_log_failure_event(kwargs, None, start_time, end_time) + + # Verify payload was added to queue + assert len(logger.log_queue) == 1 + + # Verify the payload has correct failure characteristics according to DD LLM Obs API spec + payload = logger.log_queue[0] + assert payload["trace_id"] == "test-trace-id-failure-456" + assert payload["meta"]["metadata"]["id"] == "test-request-id-failure-789" + assert payload["status"] == "error" + + # Verify error information follows DD LLM Obs API spec + assert payload["meta"]["error"]["message"] == "RateLimitError: You exceeded your current quota" + assert payload["meta"]["error"]["type"] == "RateLimitError" + assert payload["meta"]["error"]["stack"] == "Traceback (most recent call last):\n File test.py, line 1\n RateLimitError: You exceeded your current quota" + + assert payload["metrics"]["total_cost"] == 0.0 + assert payload["metrics"]["total_tokens"] == 0 + assert payload["metrics"]["output_tokens"] == 0 + + # Verify batch sending not triggered (queue size < batch_size) + mock_send_batch.assert_not_called() -class TestDataDogLLMObsLogger(DataDogLLMObsLogger): + +class TestDataDogLLMObsLoggerForRedaction(DataDogLLMObsLogger): """Test suite for DataDog LLM Observability Logger""" def __init__(self, **kwargs): super().__init__(**kwargs)