litellm/litellm/integrations/opentelemetry.py
2024-06-02 19:49:34 +09:00

114 lines
4.2 KiB
Python

from dataclasses import dataclass, field
from typing import Optional
import os
from litellm.integrations.custom_logger import CustomLogger
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.export import (
BatchSpanProcessor,
ConsoleSpanExporter,
)
LITELLM_TRACER_NAME = "litellm"
LITELLM_RESOURCE = {"service.name": "litellm"}
@dataclass
class OpenTelemetryConfig:
exporter: str = field(default="console")
endpoint: Optional[str] = None
bearer_token: Optional[str] = None
@classmethod
def from_env(cls):
return cls(
exporter=os.getenv("OTEL_EXPORTER", "console"),
endpoint=os.getenv("OTEL_ENDPOINT"),
bearer_token=os.getenv("OTEL_BEARER_TOKEN"),
)
class OpenTelemetry(CustomLogger):
def __init__(self, config=OpenTelemetryConfig.from_env()):
self.config = config
provider = TracerProvider(resource=Resource(attributes=LITELLM_RESOURCE))
provider.add_span_processor(self._get_span_processor())
trace.set_tracer_provider(provider)
self.tracer = trace.get_tracer(LITELLM_TRACER_NAME)
def log_success_event(self, kwargs, response_obj, start_time, end_time):
self._handle_sucess(kwargs, response_obj, start_time, end_time)
def log_failure_event(self, kwargs, response_obj, start_time, end_time):
self._handle_failure(kwargs, response_obj, start_time, end_time)
async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
self._handle_sucess(kwargs, response_obj, start_time, end_time)
async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
self._handle_failure(kwargs, response_obj, start_time, end_time)
def _handle_sucess(self, kwargs, response_obj, start_time, end_time):
span = self.tracer.start_span(
name=self._get_span_name(kwargs),
start_time=self._to_ns(start_time),
context=self._get_span_context(kwargs),
)
span.set_status(Status(StatusCode.OK))
self.set_attributes(span, kwargs, response_obj)
span.end(end_time=self._to_ns(end_time))
def _handle_failure(self, kwargs, response_obj, start_time, end_time):
span = self.tracer.start_span(
name=self._get_span_name(kwargs),
start_time=self._to_ns(start_time),
context=self._get_span_context(kwargs),
)
span.set_status(Status(StatusCode.ERROR))
self.set_attributes(span, kwargs, response_obj)
span.end(end_time=self._to_ns(end_time))
def set_attributes(self, span, kwargs, response_obj):
for key in ["model", "api_base", "api_version"]:
if key in kwargs:
span.set_attribute(key, kwargs[key])
def _to_ns(self, dt):
return int(dt.timestamp() * 1e9)
def _get_span_name(self, kwargs):
f"litellm-{kwargs.get('call_type', 'completion')}"
def _get_span_context(self, kwargs):
litellm_params = kwargs.get("litellm_params", {}) or {}
proxy_server_request = litellm_params.get("proxy_server_request", {}) or {}
headers = proxy_server_request.get("headers", {}) or {}
traceparent = headers.get("traceparent", None)
if traceparent is None:
return None
else:
carrier = {"traceparent": traceparent}
return TraceContextTextMapPropagator().extract(carrier=carrier)
def _get_span_processor(self):
if self.config.exporter == "console":
return BatchSpanProcessor(ConsoleSpanExporter())
elif self.config.exporter == "otlp_http":
return BatchSpanProcessor(
OTLPSpanExporter(
endpoint=self.OTEL_ENDPOINT,
headers={"Authorization": f"Bearer {self.OTEL_BEARER_TOKEN}"},
)
)
else:
return BatchSpanProcessor(ConsoleSpanExporter())