scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-10 22:41:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 83

Fix test

Browse source
This commit is contained in:
Sameer Kankute 2026-03-27 21:17:49 +05:30
parent d1139f8514
commit 2dd3ad3262
No known key found for this signature in database
1 changed files with 23 additions and 26 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

49
litellm/integrations/opentelemetry.py
View file

@ -1,4 +1,5 @@
import os
import types
from dataclasses import dataclass
from datetime import datetime
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union, cast
@ -2089,36 +2090,32 @@ class OpenTelemetry(CustomLogger):
return BatchSpanProcessor(ConsoleSpanExporter())
def _wrap_exporter_with_failure_tracking(self, exporter: SpanExporter) -> Any:
if getattr(exporter, "_litellm_failure_tracking_wrapped", False):
return exporter
original_export = exporter.export
parent = self
class _FailureTrackingExporter:
def __init__(self, wrapped_exporter: SpanExporter):
self._wrapped_exporter = wrapped_exporter
def _export_with_failure_tracking(self, spans):
try:
result = original_export(spans)
except Exception:
parent._report_span_export_failure()
raise
def export(self, spans):
try:
result = self._wrapped_exporter.export(spans)
except Exception:
parent._report_span_export_failure()
raise
if parent._is_failed_span_export_result(result):
parent._report_span_export_failure()
else:
parent._mark_span_export_success()
return result
if parent._is_failed_span_export_result(result):
parent._report_span_export_failure()
else:
parent._mark_span_export_success()
return result
def shutdown(self):
if hasattr(self._wrapped_exporter, "shutdown"):
return self._wrapped_exporter.shutdown()
return None
def force_flush(self, timeout_millis: int = 30000):
if hasattr(self._wrapped_exporter, "force_flush"):
return self._wrapped_exporter.force_flush(timeout_millis)
return True
return _FailureTrackingExporter(exporter)
setattr(
exporter,
"export",
types.MethodType(_export_with_failure_tracking, exporter),
)
setattr(exporter, "_litellm_failure_tracking_wrapped", True)
return exporter
@staticmethod
def _is_failed_span_export_result(result: Any) -> bool: