tiennm99/litellm
1
0
Fork 0
mirror of https://github.com/tiennm99/litellm.git synced 2026-08-22 08:24:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

[Feat] Datadog LLM Observability - Add support for Failure Logging (#13726)

Browse Source 
* add async_log_failure_event for DD LLM Obs

* update types

* DataDogLLMObsLogger  add failure logging support

* test_async_log_failure_event

* dd test failure
This commit is contained in:
Ishaan Jaff
2025-08-18 15:19:48 -07:00
committed by GitHub
parent 169a17400f
commit ef08e18c66
4 changed files with 186 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
litellm/integrations/datadog/datadog_llm_obs.py
+59 -2
View File
@@ -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"]:
litellm/proxy/proxy_config.yaml
+2
View File
@@ -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"]
litellm/types/integrations/datadog_llm_obs.py
+13 -4
View File
@@ -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):
tests/test_litellm/integrations/datadog/test_datadog_llm_observability.py
+112 -3
View File
@@ -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)