diff --git a/litellm/__init__.py b/litellm/__init__.py index e17ab613dac..6f1df057769 100644 --- a/litellm/__init__.py +++ b/litellm/__init__.py @@ -50,6 +50,7 @@ from litellm.types.integrations.datadog import DatadogInitParams from litellm.types.integrations.newrelic import NewRelicInitParams from litellm.litellm_core_utils.core_helpers import drop_params_env_flag from litellm.types.integrations.pointfive import PointFiveInitParams +from litellm.types.integrations.zerobus import ZerobusInitParams from litellm._logging import ( set_verbose, _turn_on_debug, @@ -157,6 +158,7 @@ _custom_logger_compatible_callbacks_literal = Literal[ "deepeval", "s3_v2", "pointfive", + "zerobus", "aws_sqs", "vector_store_pre_call_hook", "dotprompt", @@ -441,6 +443,7 @@ datadog_llm_observability_params: Optional[Union[DatadogLLMObsInitParams, Dict]] datadog_params: Optional[Union[DatadogInitParams, Dict]] = None newrelic_params: Optional[Union[NewRelicInitParams, Dict]] = None pointfive_params: Optional[Union[PointFiveInitParams, Mapping[str, object]]] = None +zerobus_params: Optional[Union[ZerobusInitParams, Mapping[str, object]]] = None aws_sqs_callback_params: Optional[Dict] = None generic_logger_headers: Optional[Dict] = None default_key_generate_params: Optional[Dict] = None diff --git a/litellm/integrations/callback_configs.json b/litellm/integrations/callback_configs.json index 6806188c97c..3dd3bcf8c65 100644 --- a/litellm/integrations/callback_configs.json +++ b/litellm/integrations/callback_configs.json @@ -399,6 +399,45 @@ }, "description": "PointFive Logging Integration" }, + { + "id": "zerobus", + "displayName": "Databricks Zerobus", + "logo": "databricks.svg", + "supports_key_team_logging": false, + "dynamic_params": { + "ZEROBUS_WORKSPACE_URL": { + "type": "text", + "ui_name": "Workspace URL", + "description": "Databricks workspace URL, e.g. https://dbc-a1b2c3d4-e5f6.cloud.databricks.com", + "required": true + }, + "ZEROBUS_SERVER_ENDPOINT": { + "type": "text", + "ui_name": "Zerobus Endpoint", + "description": "Zerobus ingest endpoint, e.g. https://.zerobus..cloud.databricks.com", + "required": true + }, + "ZEROBUS_CLIENT_ID": { + "type": "text", + "ui_name": "Service Principal Client ID", + "description": "OAuth client id of a service principal with USE CATALOG, USE SCHEMA, SELECT and MODIFY on the table", + "required": true + }, + "ZEROBUS_CLIENT_SECRET": { + "type": "password", + "ui_name": "Service Principal Client Secret", + "description": "OAuth client secret of the service principal", + "required": true + }, + "ZEROBUS_TABLE_NAME": { + "type": "text", + "ui_name": "Table", + "description": "Fully qualified Unity Catalog table, catalog.schema.table, created with the LiteLLM trace schema", + "required": true + } + }, + "description": "Databricks Zerobus Ingest Logging Integration" + }, { "id": "s3", "displayName": "S3", diff --git a/litellm/integrations/zerobus/__init__.py b/litellm/integrations/zerobus/__init__.py new file mode 100644 index 00000000000..b1f5bc2ca40 --- /dev/null +++ b/litellm/integrations/zerobus/__init__.py @@ -0,0 +1,5 @@ +"""Databricks Zerobus logging integration for LiteLLM.""" + +from litellm.integrations.zerobus.logger import ZerobusLogger + +__all__ = ("ZerobusLogger",) diff --git a/litellm/integrations/zerobus/client.py b/litellm/integrations/zerobus/client.py new file mode 100644 index 00000000000..bf3a9e3e269 --- /dev/null +++ b/litellm/integrations/zerobus/client.py @@ -0,0 +1,161 @@ +""" +Writes rows to a Unity Catalog table through the Zerobus Ingest REST API. + +Zerobus only accepts a Databricks OAuth token minted for its own resource and scoped to +the target table's privileges, so the client mints that token itself with the service +principal's client credentials and reuses it until shortly before it expires. +""" + +import asyncio +import base64 +import json +import time +from collections.abc import Callable, Mapping, Sequence +from typing import Final + +import httpx +from pydantic import BaseModel, ValidationError + +import litellm +from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler +from litellm.types.integrations.zerobus import ( + RETRYABLE_INGEST_STATUS_CODES, + TOKEN_REFRESH_LEEWAY_SECONDS, + ZerobusAccessToken, + ZerobusConnection, + ZerobusIngestFailure, +) + +TOKEN_PATH: Final = "/oidc/v1/token" +OAUTH_SCOPE: Final = "all-apis" + + +class _TokenResponse(BaseModel): + access_token: str + expires_in: float = 3600 + + +class ZerobusIngestError(Exception): + """A batch could not be written and the failure is worth retrying.""" + + +def zerobus_resource(workspace_id: str) -> str: + return f"api://databricks/workspaces/{workspace_id}/zerobusDirectWriteApi" + + +def authorization_details(table_name: str) -> str: + """The Unity Catalog privileges Zerobus requires the token to carry, as the token endpoint expects them.""" + catalog, schema, _table = table_name.split(".", 2) + return json.dumps( + ( + { + "type": "unity_catalog_privileges", + "privileges": ("USE CATALOG",), + "object_type": "CATALOG", + "object_full_path": catalog, + }, + { + "type": "unity_catalog_privileges", + "privileges": ("USE SCHEMA",), + "object_type": "SCHEMA", + "object_full_path": f"{catalog}.{schema}", + }, + { + "type": "unity_catalog_privileges", + "privileges": ("SELECT", "MODIFY"), + "object_type": "TABLE", + "object_full_path": table_name, + }, + ) + ) + + +def insert_url(connection: ZerobusConnection) -> str: + return f"{connection.server_endpoint.rstrip('/')}/zerobus/v1/tables/{connection.table_name}/insert" + + +def token_url(connection: ZerobusConnection) -> str: + return f"{connection.workspace_url.rstrip('/')}{TOKEN_PATH}" + + +def _basic_auth(client_id: str, client_secret: str) -> str: + return "Basic " + base64.b64encode(f"{client_id}:{client_secret}".encode()).decode() + + +def _status_failure(what: str, error: httpx.HTTPStatusError) -> ZerobusIngestFailure: + status: Final = error.response.status_code + return ZerobusIngestFailure( + detail=f"{what} returned {status}: {error.response.text}"[:500], + retryable=status in RETRYABLE_INGEST_STATUS_CODES, + ) + + +class ZerobusIngestClient: + def __init__( + self, + connection: ZerobusConnection, + http_client: AsyncHTTPHandler, + clock: Callable[[], float] = time.time, + ) -> None: + self.connection: Final = connection + self.http_client: Final = http_client + self.clock: Final = clock + self._token: ZerobusAccessToken | None = None + self._token_lock: Final = asyncio.Lock() + + async def insert(self, rows: Sequence[Mapping[str, object]]) -> ZerobusIngestFailure | None: + """Write ``rows`` as one request. ``None`` means Zerobus accepted every row.""" + token: Final = await self.access_token() + if isinstance(token, ZerobusIngestFailure): + return token + try: + await self.http_client.post( + insert_url(self.connection), + content=json.dumps([dict(row) for row in rows]).encode(), + headers={"Content-Type": "application/json", "Authorization": f"Bearer {token.value}"}, + ) + except httpx.HTTPStatusError as error: + if error.response.status_code == 401: + self._token = None + return ZerobusIngestFailure(detail="insert returned 401, token discarded", retryable=True) + return _status_failure("insert", error) + except (httpx.HTTPError, litellm.Timeout) as error: + return ZerobusIngestFailure(detail=f"insert failed: {error}", retryable=True) + return None + + async def access_token(self) -> ZerobusAccessToken | ZerobusIngestFailure: + """The cached token while it has more than the leeway left, otherwise a fresh one.""" + async with self._token_lock: + cached: Final = self._token + if cached is not None and cached.expires_at - self.clock() > TOKEN_REFRESH_LEEWAY_SECONDS: + return cached + minted: Final = await self._mint_token() + if isinstance(minted, ZerobusAccessToken): + self._token = minted + return minted + + async def _mint_token(self) -> ZerobusAccessToken | ZerobusIngestFailure: + connection: Final = self.connection + try: + response: Final = await self.http_client.post( + token_url(connection), + data={ + "grant_type": "client_credentials", + "scope": OAUTH_SCOPE, + "resource": zerobus_resource(connection.workspace_id), + "authorization_details": authorization_details(connection.table_name), + }, + headers={ + "Content-Type": "application/x-www-form-urlencoded", + "Authorization": _basic_auth(connection.client_id, connection.client_secret), + }, + ) + except httpx.HTTPStatusError as error: + return _status_failure("token request", error) + except (httpx.HTTPError, litellm.Timeout) as error: + return ZerobusIngestFailure(detail=f"token request failed: {error}", retryable=True) + try: + parsed: Final = _TokenResponse.model_validate_json(response.text) + except ValidationError as error: + return ZerobusIngestFailure(detail=f"token response was not understood: {error}", retryable=False) + return ZerobusAccessToken(value=parsed.access_token, expires_at=self.clock() + parsed.expires_in) diff --git a/litellm/integrations/zerobus/logger.py b/litellm/integrations/zerobus/logger.py new file mode 100644 index 00000000000..b43610518dc --- /dev/null +++ b/litellm/integrations/zerobus/logger.py @@ -0,0 +1,238 @@ +""" +Databricks Zerobus logging integration. + +Buffers one ``TRACE_TABLE_COLUMNS`` row per request and writes each flush to a Unity +Catalog Delta table through the Zerobus Ingest REST API. +""" + +import asyncio +from collections.abc import Mapping +from datetime import datetime +from typing import Final +from urllib.parse import urlsplit + +import litellm +from litellm._logging import verbose_logger +from litellm.integrations.custom_batch_logger import CustomBatchLogger +from litellm.integrations.zerobus.client import ZerobusIngestClient, ZerobusIngestError +from litellm.integrations.zerobus.row import trace_row +from litellm.litellm_core_utils.redact_messages import ( + redacted_standard_logging_payload, + should_redact_message_logging, +) +from litellm.llms.custom_httpx.http_handler import get_async_httpx_client, httpxSpecialProvider +from litellm.secret_managers.main import get_secret_str +from litellm.types.integrations.zerobus import ZerobusConnection, ZerobusInitParams + +_ENV_REFERENCE_PREFIX: Final = "os.environ/" + + +def _resolved_secret(value: str | None) -> str | None: + """Resolve a config value that may name a secret; an unset ``os.environ/NAME`` stays unresolved.""" + if value is None: + return None + resolved: Final = get_secret_str(value) + if resolved: + return resolved + return None if value.startswith(_ENV_REFERENCE_PREFIX) else value + + +def _configured_params() -> ZerobusInitParams: + configured: Final = litellm.zerobus_params + if isinstance(configured, ZerobusInitParams): + return configured + if isinstance(configured, Mapping): + return ZerobusInitParams.model_validate(configured) + return ZerobusInitParams() + + +def _setting(configured: str | None, env_var: str) -> str: + """Prefer the configured value, falling back to the environment the proxy UI writes.""" + value: Final = _resolved_secret(configured) or get_secret_str(env_var) + if not value: + raise ValueError( + f"zerobus logging requires {env_var}. Set it in the environment, or " + f"litellm_settings.zerobus_params.{env_var.removeprefix('ZEROBUS_').lower()} in config.yaml" + ) + return value + + +def _workspace_id(server_endpoint: str) -> str: + """The Zerobus endpoint is ``https://.zerobus..``, so the id is its first label.""" + host: Final = urlsplit(server_endpoint).hostname or "" + workspace_id: Final = host.split(".", 1)[0] + if not workspace_id.isdigit(): + raise ValueError( + f"ZEROBUS_SERVER_ENDPOINT {server_endpoint!r} does not look like " + "https://.zerobus..cloud.databricks.com" + ) + return workspace_id + + +def _table_name(configured: str | None) -> str: + table_name: Final = _setting(configured, "ZEROBUS_TABLE_NAME") + if table_name.count(".") != 2: + raise ValueError(f"ZEROBUS_TABLE_NAME {table_name!r} must be fully qualified as catalog.schema.table") + return table_name + + +def connection_for(params: ZerobusInitParams) -> ZerobusConnection: + """The connection configured right now, so a UI edit takes effect without a restart.""" + server_endpoint: Final = _setting(params.server_endpoint, "ZEROBUS_SERVER_ENDPOINT") + return ZerobusConnection( + workspace_url=_setting(params.workspace_url, "ZEROBUS_WORKSPACE_URL"), + workspace_id=_workspace_id(server_endpoint), + server_endpoint=server_endpoint, + client_id=_setting(params.client_id, "ZEROBUS_CLIENT_ID"), + client_secret=_setting(params.client_secret, "ZEROBUS_CLIENT_SECRET"), + table_name=_table_name(params.table_name), + ) + + +class ZerobusLogger(CustomBatchLogger): + """Batching callback that writes LiteLLM request logs to a Databricks Delta table.""" + + preserve_events_added_during_flush = True + + def __init__( + self, + params: ZerobusInitParams | None = None, + client: ZerobusIngestClient | None = None, + start_periodic_flush: bool = True, + ) -> None: + resolved: Final = params if params is not None else _configured_params() + self.params: Final = resolved + self.given_client: Final = client + self._cached_client: ZerobusIngestClient | None = None + if client is None: + connection_for(resolved) + super().__init__( + flush_lock=asyncio.Lock(), + batch_size=resolved.batch_size, + flush_interval=resolved.flush_interval, + turn_off_message_logging=bool(resolved.turn_off_message_logging), + ) + self._flushing: bool = False + self._batch_flush_task: asyncio.Task[None] | None = None + self._periodic_flush_task: asyncio.Task[None] | None = ( + self._start_periodic_flush_task() if start_periodic_flush else None + ) + + @property + def client(self) -> ZerobusIngestClient: + """A client for the current connection, kept while the connection is unchanged so its token is reused.""" + if self.given_client is not None: + return self.given_client + connection: Final = connection_for(self.params) + cached: Final = self._cached_client + if cached is not None and cached.connection == connection: + return cached + fresh: Final = ZerobusIngestClient( + connection=connection, + http_client=get_async_httpx_client(llm_provider=httpxSpecialProvider.LoggingCallback), + ) + self._cached_client = fresh + return fresh + + def _start_periodic_flush_task(self) -> asyncio.Task[None] | None: + try: + loop: Final = asyncio.get_running_loop() + except RuntimeError: + return None + return loop.create_task(self.periodic_flush()) + + def _start_batch_flush_task(self) -> None: + if self._batch_flush_task is not None and not self._batch_flush_task.done(): + return + try: + loop: Final = asyncio.get_running_loop() + except RuntimeError: + return + self._batch_flush_task = loop.create_task(self.flush_queue(skip_if_flushing=True)) + + def _flush_task_is_alive(self) -> bool: + task: Final = self._periodic_flush_task + return task is not None and not task.done() and not task.get_loop().is_closed() + + async def async_log_success_event( + self, + kwargs: Mapping[str, object], + response_obj: object, + start_time: datetime, + end_time: datetime, + ) -> None: + await self._enqueue(kwargs) + + async def async_log_failure_event( + self, + kwargs: Mapping[str, object], + response_obj: object, + start_time: datetime, + end_time: datetime, + ) -> None: + await self._enqueue(kwargs) + + async def _enqueue(self, kwargs: Mapping[str, object]) -> None: + try: + if not self._flush_task_is_alive(): + self._periodic_flush_task = self._start_periodic_flush_task() + + payload: Final = self._payload_for(kwargs) + if payload is None: + verbose_logger.debug("zerobus: event carried no standard_logging_object, skipping") + return + + self.log_queue.append(trace_row(payload)) + self._drop_overflow() + if len(self.log_queue) >= self.batch_size: + self._start_batch_flush_task() + except Exception: # noqa: BLE001 # logging must never break the request path + verbose_logger.exception("zerobus: failed to queue an event") + + def _payload_for(self, kwargs: Mapping[str, object]) -> Mapping[str, object] | None: + """The payload to buffer, redacted the way the framework redacts the success path.""" + details: Final = self.redact_standard_logging_payload_from_model_call_details( + dict(kwargs) # mutable-ok: both framework helpers take the call details as a dict + ) + payload: Final = details.get("standard_logging_object") + if not isinstance(payload, dict): + return None + if should_redact_message_logging(details): + return redacted_standard_logging_payload(payload) + return payload + + def _drop_overflow(self) -> None: + if self._flushing: + return + overflow: Final = len(self.log_queue) - self.max_queue_size + if overflow <= 0: + return + del self.log_queue[:overflow] + verbose_logger.warning("zerobus: queue over %s rows, dropped %s oldest", self.max_queue_size, overflow) + + async def flush_queue(self, skip_if_flushing: bool = False) -> None: + if skip_if_flushing and self._flushing: + return + self._flushing = True + try: + await super().flush_queue() + finally: + self._flushing = False + + async def async_send_batch(self) -> None: + """ + Write everything queued as one insert. + + A retryable failure propagates so ``CustomBatchLogger`` keeps the rows for the next + flush. A failure Zerobus would repeat, a schema mismatch for one, drops the batch, + since holding it would block every row queued behind it. + """ + rows: Final = tuple(self.log_queue) + if not rows: + return + failure: Final = await self.client.insert(rows) + if failure is None: + return + if failure.retryable: + raise ZerobusIngestError(failure.detail) + verbose_logger.error("zerobus: dropping %s rows, %s", len(rows), failure.detail) diff --git a/litellm/integrations/zerobus/row.py b/litellm/integrations/zerobus/row.py new file mode 100644 index 00000000000..c4da7975c44 --- /dev/null +++ b/litellm/integrations/zerobus/row.py @@ -0,0 +1,156 @@ +""" +Shape of one Delta table row per LiteLLM request. + +Zerobus validates every record against the target table and rejects unknown columns, so +the row is a fixed set of scalar columns for filtering plus JSON-encoded ``VARIANT`` +columns for anything nested. ``create_table_sql`` renders the matching DDL. +""" + +from collections.abc import Mapping +from types import MappingProxyType +from typing import Final + +from litellm.litellm_core_utils.safe_json_dumps import safe_dumps + +TRACE_TABLE_COLUMNS: Final[Mapping[str, str]] = MappingProxyType( + { + "id": "STRING", + "trace_id": "STRING", + "session_id": "STRING", + "litellm_call_id": "STRING", + "call_type": "STRING", + "status": "STRING", + "model": "STRING", + "model_group": "STRING", + "model_id": "STRING", + "custom_llm_provider": "STRING", + "api_base": "STRING", + "stream": "BOOLEAN", + "cache_hit": "BOOLEAN", + "start_time": "TIMESTAMP", + "end_time": "TIMESTAMP", + "completion_start_time": "TIMESTAMP", + "response_time": "DOUBLE", + "prompt_tokens": "LONG", + "completion_tokens": "LONG", + "total_tokens": "LONG", + "response_cost": "DOUBLE", + "saved_cache_cost": "DOUBLE", + "api_key_hash": "STRING", + "api_key_alias": "STRING", + "team_id": "STRING", + "team_alias": "STRING", + "user_id": "STRING", + "org_id": "STRING", + "end_user": "STRING", + "requester_ip_address": "STRING", + "user_agent": "STRING", + "request_tags": "VARIANT", + "messages": "VARIANT", + "response": "VARIANT", + "error_str": "STRING", + "error_information": "VARIANT", + "metadata": "VARIANT", + "model_parameters": "VARIANT", + "hidden_params": "VARIANT", + "guardrail_information": "VARIANT", + "cost_breakdown": "VARIANT", + } +) + +_MICROSECONDS: Final = 1_000_000 + + +def create_table_sql(table_name: str) -> str: + columns: Final = ",\n".join(f" {name} {delta_type}" for name, delta_type in TRACE_TABLE_COLUMNS.items()) + return f"CREATE TABLE {table_name} (\n{columns}\n);" + + +def _text(payload: Mapping[str, object], key: str) -> str | None: + value: Final = payload.get(key) + return value if isinstance(value, str) else None + + +def _flag(payload: Mapping[str, object], key: str) -> bool | None: + value: Final = payload.get(key) + return value if isinstance(value, bool) else None + + +def _number(payload: Mapping[str, object], key: str) -> float | None: + value: Final = payload.get(key) + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + return float(value) + + +def _count(payload: Mapping[str, object], key: str) -> int | None: + value: Final = _number(payload, key) + return None if value is None else int(value) + + +def _timestamp_micros(payload: Mapping[str, object], key: str) -> int | None: + """Delta ``TIMESTAMP`` over Zerobus is epoch microseconds; LiteLLM keeps epoch seconds.""" + seconds: Final = _number(payload, key) + if seconds is None or seconds <= 0: + return None + return int(seconds * _MICROSECONDS) + + +def _json(payload: Mapping[str, object], key: str) -> str | None: + value: Final = payload.get(key) + return None if value is None else safe_dumps(value) + + +def _metadata(payload: Mapping[str, object]) -> Mapping[str, object]: + value: Final = payload.get("metadata") + return value if isinstance(value, Mapping) else MappingProxyType({}) + + +def trace_row(payload: Mapping[str, object]) -> Mapping[str, object]: + """One ``TRACE_TABLE_COLUMNS`` row for a ``StandardLoggingPayload``.""" + metadata: Final = _metadata(payload) + return MappingProxyType( + { + "id": _text(payload, "id"), + "trace_id": _text(payload, "trace_id"), + "session_id": _text(payload, "session_id"), + "litellm_call_id": _text(payload, "litellm_call_id"), + "call_type": _text(payload, "call_type"), + "status": _text(payload, "status"), + "model": _text(payload, "model"), + "model_group": _text(payload, "model_group"), + "model_id": _text(payload, "model_id"), + "custom_llm_provider": _text(payload, "custom_llm_provider"), + "api_base": _text(payload, "api_base"), + "stream": _flag(payload, "stream"), + "cache_hit": _flag(payload, "cache_hit"), + "start_time": _timestamp_micros(payload, "startTime"), + "end_time": _timestamp_micros(payload, "endTime"), + "completion_start_time": _timestamp_micros(payload, "completionStartTime"), + "response_time": _number(payload, "response_time"), + "prompt_tokens": _count(payload, "prompt_tokens"), + "completion_tokens": _count(payload, "completion_tokens"), + "total_tokens": _count(payload, "total_tokens"), + "response_cost": _number(payload, "response_cost"), + "saved_cache_cost": _number(payload, "saved_cache_cost"), + "api_key_hash": _text(metadata, "user_api_key_hash"), + "api_key_alias": _text(metadata, "user_api_key_alias"), + "team_id": _text(metadata, "user_api_key_team_id"), + "team_alias": _text(metadata, "user_api_key_team_alias"), + "user_id": _text(metadata, "user_api_key_user_id"), + "org_id": _text(metadata, "user_api_key_org_id"), + "end_user": _text(payload, "end_user"), + "requester_ip_address": _text(payload, "requester_ip_address"), + "user_agent": _text(payload, "user_agent"), + "request_tags": _json(payload, "request_tags"), + "messages": _json(payload, "messages"), + "response": _json(payload, "response"), + "error_str": _text(payload, "error_str"), + "error_information": _json(payload, "error_information"), + "metadata": _json(payload, "metadata"), + "model_parameters": _json(payload, "model_parameters"), + "hidden_params": _json(payload, "hidden_params"), + "guardrail_information": _json(payload, "guardrail_information"), + "cost_breakdown": _json(payload, "cost_breakdown"), + } + ) diff --git a/litellm/litellm_core_utils/custom_logger_registry.py b/litellm/litellm_core_utils/custom_logger_registry.py index 6294f3bc577..7049fdd1f39 100644 --- a/litellm/litellm_core_utils/custom_logger_registry.py +++ b/litellm/litellm_core_utils/custom_logger_registry.py @@ -52,6 +52,7 @@ from litellm.integrations.vantage.vantage_logger import VantageLogger from litellm.integrations.vector_store_integrations.vector_store_pre_call_hook import ( VectorStorePreCallHook, ) +from litellm.integrations.zerobus import ZerobusLogger from litellm.proxy.hooks.dynamic_rate_limiter import _PROXY_DynamicRateLimitHandler from litellm.proxy.hooks.dynamic_rate_limiter_v3 import _PROXY_DynamicRateLimitHandlerV3 @@ -97,6 +98,7 @@ class CustomLoggerRegistry: "deepeval": DeepEvalLogger, "s3_v2": S3Logger, "pointfive": PointFiveLogger, + "zerobus": ZerobusLogger, "aws_sqs": SQSLogger, "dynamic_rate_limiter": _PROXY_DynamicRateLimitHandler, "dynamic_rate_limiter_v3": _PROXY_DynamicRateLimitHandlerV3, diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index f7679b31f69..6e0d7929a43 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -196,6 +196,7 @@ from ..integrations.s3 import S3Logger from ..integrations.s3_v2 import S3Logger as S3V2Logger from ..integrations.supabase import Supabase from ..integrations.traceloop import TraceloopLogger +from ..integrations.zerobus import ZerobusLogger from .exception_mapping_utils import _get_response_headers from .initialize_dynamic_callback_params import ( get_trusted_callback_params, @@ -4433,6 +4434,14 @@ def _init_custom_logger_compatible_class( _pointfive_logger: Final = PointFiveLogger() _in_memory_loggers.append(_pointfive_logger) return _pointfive_logger + elif logging_integration == "zerobus": + for callback in _in_memory_loggers: + if isinstance(callback, ZerobusLogger): + return callback + + _zerobus_logger: Final = ZerobusLogger() + _in_memory_loggers.append(_zerobus_logger) + return _zerobus_logger elif logging_integration == "aws_sqs": for callback in _in_memory_loggers: if isinstance(callback, SQSLogger): @@ -5125,6 +5134,10 @@ def get_custom_logger_compatible_class( for callback in _in_memory_loggers: if isinstance(callback, PointFiveLogger): return callback + elif logging_integration == "zerobus": + for callback in _in_memory_loggers: + if isinstance(callback, ZerobusLogger): + return callback elif logging_integration == "aws_sqs": for callback in _in_memory_loggers: if isinstance(callback, SQSLogger): diff --git a/litellm/proxy/_types.py b/litellm/proxy/_types.py index 6322a1212fe..a28820bf229 100644 --- a/litellm/proxy/_types.py +++ b/litellm/proxy/_types.py @@ -3925,6 +3925,18 @@ class AllCallbacks(LiteLLMPydanticObjectBase): ], ) + zerobus: CallbackOnUI = CallbackOnUI( + litellm_callback_name="zerobus", + ui_callback_name="Databricks Zerobus", + litellm_callback_params=[ # mutable-ok: the registry field is typed list + "ZEROBUS_WORKSPACE_URL", + "ZEROBUS_SERVER_ENDPOINT", + "ZEROBUS_CLIENT_ID", + "ZEROBUS_CLIENT_SECRET", + "ZEROBUS_TABLE_NAME", + ], + ) + class SpendLogsRouterMetadata(TypedDict): """ diff --git a/litellm/types/integrations/zerobus.py b/litellm/types/integrations/zerobus.py new file mode 100644 index 00000000000..6f23cf88055 --- /dev/null +++ b/litellm/types/integrations/zerobus.py @@ -0,0 +1,53 @@ +from dataclasses import dataclass +from typing import Final + +from pydantic import Field + +from litellm.types.integrations.custom_logger import StandardCustomLoggerInitParams + +RETRYABLE_INGEST_STATUS_CODES: Final = frozenset({408, 429, 500, 502, 503, 504}) + +TOKEN_REFRESH_LEEWAY_SECONDS: Final = 60 + + +class ZerobusInitParams(StandardCustomLoggerInitParams): + """ + Params for initializing a Databricks Zerobus logger on litellm. + + Every connection field falls back to its ``ZEROBUS_*`` environment variable, which is + what the proxy UI writes. ``table_name`` is the fully qualified ``catalog.schema.table``. + """ + + workspace_url: str | None = None + server_endpoint: str | None = None + client_id: str | None = None + client_secret: str | None = None + table_name: str | None = None + batch_size: int = Field(default=100, gt=0) + flush_interval: int = Field(default=10, gt=0) + + +@dataclass(frozen=True, slots=True) +class ZerobusConnection: + """Everything needed to mint a token for one table and post rows to it.""" + + workspace_url: str + workspace_id: str + server_endpoint: str + client_id: str + client_secret: str + table_name: str + + +@dataclass(frozen=True, slots=True) +class ZerobusAccessToken: + value: str + expires_at: float + + +@dataclass(frozen=True, slots=True) +class ZerobusIngestFailure: + """Why a batch could not be written, and whether a later attempt could still succeed.""" + + detail: str + retryable: bool diff --git a/tests/test_litellm/integrations/zerobus/test_client.py b/tests/test_litellm/integrations/zerobus/test_client.py new file mode 100644 index 00000000000..fe21f70329e --- /dev/null +++ b/tests/test_litellm/integrations/zerobus/test_client.py @@ -0,0 +1,223 @@ +import base64 +import json +from collections.abc import Sequence + +import httpx +import pytest + +from litellm.integrations.zerobus.client import ZerobusIngestClient +from litellm.types.integrations.zerobus import ZerobusConnection, ZerobusIngestFailure + +CONNECTION = ZerobusConnection( + workspace_url="https://dbc-a1b2c3d4-e5f6.cloud.databricks.com/", + workspace_id="1234567890123456", + server_endpoint="https://1234567890123456.zerobus.us-west-2.cloud.databricks.com", + client_id="sp-client-id", + client_secret="sp-client-secret", + table_name="main.litellm.traces", +) +ROWS = ({"id": "a", "model": "gpt-4o"}, {"id": "b", "model": "gpt-4o"}) + + +def _token(value: str = "tok-1", expires_in: float = 3600) -> httpx.Response: + return httpx.Response(200, text=json.dumps({"access_token": value, "expires_in": expires_in})) + + +def _accepted() -> httpx.Response: + return httpx.Response(200, text="{}") + + +class FakeHTTPClient: + """ + Stands in for AsyncHTTPHandler, including its habit of raising on error statuses. + + Results are consumed in order, and the last one repeats. + """ + + def __init__( + self, + token: Sequence[httpx.Response | Exception] | None = None, + insert: Sequence[httpx.Response | Exception] | None = None, + ) -> None: + self.token_results = list(token) if token else [_token()] # mutable-ok: results are consumed by popping + self.insert_results = list(insert) if insert else [_accepted()] # mutable-ok: results are consumed by popping + self.token_calls: list[dict] = [] + self.insert_calls: list[dict] = [] + + async def post(self, url, data=None, content=None, headers=None, **_): + if url.endswith("/oidc/v1/token"): + self.token_calls.append({"url": url, "data": data, "headers": headers or {}}) + return _next_result(self.token_results, url) + self.insert_calls.append({"url": url, "content": content, "headers": headers or {}}) + return _next_result(self.insert_results, url) + + +def _next_result(results: list, url: str) -> httpx.Response: + result = results.pop(0) if len(results) > 1 else results[0] + if isinstance(result, Exception): + raise result + if result.status_code >= 300: + raise httpx.HTTPStatusError( + "boom", + request=httpx.Request("POST", url), + response=httpx.Response(result.status_code, text=result.text), + ) + return result + + +class FakeClock: + def __init__(self, now: float = 1_000.0) -> None: + self.now = now + + def __call__(self) -> float: + return self.now + + +def _client(http_client: FakeHTTPClient, clock: FakeClock | None = None) -> ZerobusIngestClient: + return ZerobusIngestClient(connection=CONNECTION, http_client=http_client, clock=clock or FakeClock()) + + +@pytest.mark.asyncio +async def test_rows_are_posted_as_one_json_list_to_the_table_insert_endpoint(): + http_client = FakeHTTPClient() + + outcome = await _client(http_client).insert(ROWS) + + assert outcome is None + (call,) = http_client.insert_calls + assert call["url"] == ( + "https://1234567890123456.zerobus.us-west-2.cloud.databricks.com/zerobus/v1/tables/main.litellm.traces/insert" + ) + assert json.loads(call["content"]) == [{"id": "a", "model": "gpt-4o"}, {"id": "b", "model": "gpt-4o"}] + assert call["headers"]["Content-Type"] == "application/json" + assert call["headers"]["Authorization"] == "Bearer tok-1" + + +@pytest.mark.asyncio +async def test_the_token_is_minted_for_the_zerobus_resource_with_the_table_privileges(): + """Zerobus refuses a plain workspace token: it must name its own resource and the table's UC privileges.""" + http_client = FakeHTTPClient() + + await _client(http_client).insert(ROWS) + + (call,) = http_client.token_calls + assert call["url"] == "https://dbc-a1b2c3d4-e5f6.cloud.databricks.com/oidc/v1/token" + assert call["data"]["grant_type"] == "client_credentials" + assert call["data"]["scope"] == "all-apis" + assert call["data"]["resource"] == "api://databricks/workspaces/1234567890123456/zerobusDirectWriteApi" + details = json.loads(call["data"]["authorization_details"]) + assert [(d["object_type"], d["object_full_path"], d["privileges"]) for d in details] == [ + ("CATALOG", "main", ["USE CATALOG"]), + ("SCHEMA", "main.litellm", ["USE SCHEMA"]), + ("TABLE", "main.litellm.traces", ["SELECT", "MODIFY"]), + ] + assert all(d["type"] == "unity_catalog_privileges" for d in details) + + +@pytest.mark.asyncio +async def test_the_service_principal_authenticates_with_http_basic(): + http_client = FakeHTTPClient() + + await _client(http_client).insert(ROWS) + + scheme, credentials = http_client.token_calls[0]["headers"]["Authorization"].split(" ") + assert scheme == "Basic" + assert base64.b64decode(credentials).decode() == "sp-client-id:sp-client-secret" + + +@pytest.mark.asyncio +async def test_the_token_is_reused_across_inserts_until_it_nears_expiry(): + clock = FakeClock(now=1_000.0) + http_client = FakeHTTPClient(token=[_token("tok-1", expires_in=600), _token("tok-2")]) + client = _client(http_client, clock) + + await client.insert(ROWS) + clock.now = 1_000.0 + 600 - 61 + await client.insert(ROWS) + clock.now = 1_000.0 + 600 - 59 + await client.insert(ROWS) + + assert len(http_client.token_calls) == 2 + assert [call["headers"]["Authorization"] for call in http_client.insert_calls] == [ + "Bearer tok-1", + "Bearer tok-1", + "Bearer tok-2", + ] + + +@pytest.mark.asyncio +async def test_a_401_discards_the_token_so_the_next_insert_mints_a_fresh_one(): + http_client = FakeHTTPClient( + token=[_token("tok-1"), _token("tok-2")], + insert=[httpx.Response(401, text="expired"), _accepted()], + ) + client = _client(http_client) + + first = await client.insert(ROWS) + second = await client.insert(ROWS) + + assert first == ZerobusIngestFailure(detail="insert returned 401, token discarded", retryable=True) + assert second is None + assert http_client.insert_calls[1]["headers"]["Authorization"] == "Bearer tok-2" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", [429, 500, 503]) +async def test_a_transient_insert_status_is_retryable(status: int): + http_client = FakeHTTPClient(insert=[httpx.Response(status, text="later")]) + + outcome = await _client(http_client).insert(ROWS) + + assert isinstance(outcome, ZerobusIngestFailure) + assert outcome.retryable is True + assert str(status) in outcome.detail + + +@pytest.mark.asyncio +async def test_a_schema_rejection_is_not_retryable_and_says_why(): + http_client = FakeHTTPClient(insert=[httpx.Response(400, text="unknown column foo")]) + + outcome = await _client(http_client).insert(ROWS) + + assert outcome == ZerobusIngestFailure(detail="insert returned 400: unknown column foo", retryable=False) + + +@pytest.mark.asyncio +async def test_a_network_failure_on_insert_is_retryable(): + http_client = FakeHTTPClient(insert=[httpx.ConnectError("connection refused")]) + + outcome = await _client(http_client).insert(ROWS) + + assert isinstance(outcome, ZerobusIngestFailure) + assert outcome.retryable is True + + +@pytest.mark.asyncio +async def test_bad_credentials_fail_the_insert_without_posting_rows(): + http_client = FakeHTTPClient(token=[httpx.Response(401, text="invalid_client")]) + + outcome = await _client(http_client).insert(ROWS) + + assert outcome == ZerobusIngestFailure(detail="token request returned 401: invalid_client", retryable=False) + assert http_client.insert_calls == [] + + +@pytest.mark.asyncio +async def test_a_token_endpoint_outage_is_retryable(): + http_client = FakeHTTPClient(token=[httpx.Response(503, text="try later")]) + + outcome = await _client(http_client).insert(ROWS) + + assert isinstance(outcome, ZerobusIngestFailure) + assert outcome.retryable is True + + +@pytest.mark.asyncio +async def test_a_token_response_without_a_token_is_reported_not_raised(): + http_client = FakeHTTPClient(token=[httpx.Response(200, text='{"token_type": "Bearer"}')]) + + outcome = await _client(http_client).insert(ROWS) + + assert isinstance(outcome, ZerobusIngestFailure) + assert outcome.retryable is False + assert "token response" in outcome.detail diff --git a/tests/test_litellm/integrations/zerobus/test_logger.py b/tests/test_litellm/integrations/zerobus/test_logger.py new file mode 100644 index 00000000000..270bccf52cb --- /dev/null +++ b/tests/test_litellm/integrations/zerobus/test_logger.py @@ -0,0 +1,328 @@ +import asyncio +from collections.abc import Callable, Mapping, Sequence + +import pytest + +import litellm +from litellm.integrations.zerobus.client import ZerobusIngestError +from litellm.integrations.zerobus.logger import ZerobusLogger, connection_for +from litellm.types.integrations.zerobus import ZerobusIngestFailure, ZerobusInitParams + +WORKSPACE_URL = "https://dbc-a1b2c3d4-e5f6.cloud.databricks.com" +SERVER_ENDPOINT = "https://1234567890123456.zerobus.us-west-2.cloud.databricks.com" + + +class FakeIngestClient: + """Records the rows each flush would have written.""" + + def __init__(self, outcomes: Sequence[ZerobusIngestFailure | None] = (None,)) -> None: + self.outcomes = list(outcomes) # mutable-ok: outcomes are consumed by popping + self.batches: list[tuple[Mapping[str, object], ...]] = [] + self.on_insert: Callable[[], None] | None = None + + async def insert(self, rows: Sequence[Mapping[str, object]]) -> ZerobusIngestFailure | None: + if self.on_insert is not None: + self.on_insert() + self.batches.append(tuple(rows)) + return self.outcomes.pop(0) if len(self.outcomes) > 1 else self.outcomes[0] + + def ids(self) -> list[object]: + return [row["id"] for batch in self.batches for row in batch] + + +def _logger(client: FakeIngestClient, **params) -> ZerobusLogger: + return ZerobusLogger(params=ZerobusInitParams(**params), client=client) + + +def _event(request_id: str, **payload) -> dict: + return { + "standard_logging_object": { + "id": request_id, + "model": "gpt-4o", + "messages": [{"role": "user", "content": "hi"}], + "response": {"choices": []}, + **payload, + } + } + + +async def _settle(logger: ZerobusLogger) -> None: + for _ in range(200): + await asyncio.sleep(0.001) + task = logger._batch_flush_task + if (task is None or task.done()) and not logger._flushing: + return + + +@pytest.mark.asyncio +async def test_a_full_batch_is_written_as_one_insert_of_table_rows(): + client = FakeIngestClient() + logger = _logger(client, batch_size=3) + + for request_id in ("a", "b", "c"): + await logger.async_log_success_event(_event(request_id), None, None, None) + + await _settle(logger) + assert len(client.batches) == 1 + assert client.ids() == ["a", "b", "c"] + assert client.batches[0][0]["model"] == "gpt-4o" + assert logger.log_queue == [] + + +@pytest.mark.asyncio +async def test_rows_are_held_until_the_batch_is_full(): + client = FakeIngestClient() + logger = _logger(client, batch_size=3) + + await logger.async_log_success_event(_event("a"), None, None, None) + + assert client.batches == [] + assert len(logger.log_queue) == 1 + + +@pytest.mark.asyncio +async def test_failed_requests_are_written_too(): + client = FakeIngestClient() + logger = _logger(client, batch_size=1) + + await logger.async_log_failure_event(_event("failed", status="failure", error_str="boom"), None, None, None) + + await _settle(logger) + assert client.ids() == ["failed"] + assert client.batches[0][0]["status"] == "failure" + assert client.batches[0][0]["error_str"] == "boom" + + +@pytest.mark.asyncio +async def test_an_event_without_a_standard_payload_is_skipped(): + client = FakeIngestClient() + logger = _logger(client, batch_size=1) + + await logger.async_log_success_event({"kwargs": "but no payload"}, None, None, None) + + assert client.batches == [] + assert logger.log_queue == [] + + +@pytest.mark.asyncio +async def test_a_retryable_failure_keeps_the_rows_for_the_next_flush(): + client = FakeIngestClient([ZerobusIngestFailure("zerobus is down", retryable=True)]) + logger = _logger(client, batch_size=2) + + for request_id in ("a", "b"): + await logger.async_log_success_event(_event(request_id), None, None, None) + + await _settle(logger) + assert [row["id"] for row in logger.log_queue] == ["a", "b"] + + +@pytest.mark.asyncio +async def test_a_retryable_failure_surfaces_so_the_base_logger_can_preserve_it(): + client = FakeIngestClient([ZerobusIngestFailure("zerobus is down", retryable=True)]) + logger = _logger(client, batch_size=99) + logger.log_queue.append({"id": "a"}) + + with pytest.raises(ZerobusIngestError, match="zerobus is down"): + await logger.async_send_batch() + + +@pytest.mark.asyncio +async def test_a_rejected_batch_is_dropped_rather_than_blocking_the_queue(): + client = FakeIngestClient([ZerobusIngestFailure("unknown column", retryable=False)]) + logger = _logger(client, batch_size=2) + + for request_id in ("a", "b"): + await logger.async_log_success_event(_event(request_id), None, None, None) + + await _settle(logger) + assert logger.log_queue == [] + + +@pytest.mark.asyncio +async def test_a_row_that_arrives_mid_flush_is_kept_for_the_next_one(): + client = FakeIngestClient() + logger = _logger(client, batch_size=1) + client.on_insert = lambda: logger.log_queue.append({"id": "late"}) + + await logger.async_log_success_event(_event("first"), None, None, None) + + await _settle(logger) + assert client.ids() == ["first"] + assert [row["id"] for row in logger.log_queue] == ["late"] + + +@pytest.mark.asyncio +async def test_a_client_error_does_not_break_the_request_path(): + class ExplodingClient: + async def insert(self, rows): + raise RuntimeError("bug") + + logger = ZerobusLogger(params=ZerobusInitParams(batch_size=1), client=ExplodingClient()) + + await logger.async_log_success_event(_event("a"), None, None, None) + await _settle(logger) + + assert [row["id"] for row in logger.log_queue] == ["a"] + + +@pytest.mark.asyncio +async def test_turn_off_message_logging_redacts_prompts_and_responses_but_keeps_the_rest(): + client = FakeIngestClient() + logger = _logger(client, batch_size=1, turn_off_message_logging=True) + + await logger.async_log_success_event( + _event("a", prompt_tokens=10, response={"choices": [{"message": {"content": "the secret answer"}}]}), + None, + None, + None, + ) + + await _settle(logger) + (row,) = client.batches[0] + assert row["id"] == "a" + assert row["prompt_tokens"] == 10 + assert '"hi"' not in str(row["messages"]) + assert "the secret answer" not in str(row["response"]) + + +def test_connection_comes_from_the_environment_the_proxy_ui_writes(monkeypatch): + monkeypatch.setenv("ZEROBUS_WORKSPACE_URL", WORKSPACE_URL) + monkeypatch.setenv("ZEROBUS_SERVER_ENDPOINT", SERVER_ENDPOINT) + monkeypatch.setenv("ZEROBUS_CLIENT_ID", "sp-id") + monkeypatch.setenv("ZEROBUS_CLIENT_SECRET", "sp-secret") + monkeypatch.setenv("ZEROBUS_TABLE_NAME", "main.litellm.traces") + + connection = connection_for(ZerobusInitParams()) + + assert connection.workspace_url == WORKSPACE_URL + assert connection.server_endpoint == SERVER_ENDPOINT + assert connection.workspace_id == "1234567890123456" + assert connection.client_id == "sp-id" + assert connection.client_secret == "sp-secret" + assert connection.table_name == "main.litellm.traces" + + +def test_config_yaml_params_win_over_the_environment(monkeypatch): + monkeypatch.setenv("ZEROBUS_TABLE_NAME", "env.schema.table") + monkeypatch.setenv("ZEROBUS_CLIENT_SECRET", "from-env") + + connection = connection_for( + ZerobusInitParams( + workspace_url=WORKSPACE_URL, + server_endpoint=SERVER_ENDPOINT, + client_id="sp-id", + client_secret="from-config", + table_name="cfg.schema.table", + ) + ) + + assert connection.table_name == "cfg.schema.table" + assert connection.client_secret == "from-config" + + +def test_a_secret_reference_in_config_yaml_is_resolved(monkeypatch): + monkeypatch.setenv("MY_SP_SECRET", "resolved-secret") + + connection = connection_for( + ZerobusInitParams( + workspace_url=WORKSPACE_URL, + server_endpoint=SERVER_ENDPOINT, + client_id="sp-id", + client_secret="os.environ/MY_SP_SECRET", + table_name="main.litellm.traces", + ) + ) + + assert connection.client_secret == "resolved-secret" + + +def test_a_missing_setting_names_the_env_var_to_set(monkeypatch): + monkeypatch.delenv("ZEROBUS_CLIENT_SECRET", raising=False) + + with pytest.raises(ValueError, match="ZEROBUS_CLIENT_SECRET"): + connection_for( + ZerobusInitParams( + workspace_url=WORKSPACE_URL, + server_endpoint=SERVER_ENDPOINT, + client_id="sp-id", + table_name="main.litellm.traces", + ) + ) + + +def test_a_table_that_is_not_fully_qualified_is_refused(): + with pytest.raises(ValueError, match="catalog.schema.table"): + connection_for( + ZerobusInitParams( + workspace_url=WORKSPACE_URL, + server_endpoint=SERVER_ENDPOINT, + client_id="sp-id", + client_secret="sp-secret", + table_name="traces", + ) + ) + + +def test_an_endpoint_without_a_workspace_id_is_refused(): + """The token's resource needs the numeric workspace id, which only the Zerobus hostname carries.""" + with pytest.raises(ValueError, match="ZEROBUS_SERVER_ENDPOINT"): + connection_for( + ZerobusInitParams( + workspace_url=WORKSPACE_URL, + server_endpoint=WORKSPACE_URL, + client_id="sp-id", + client_secret="sp-secret", + table_name="main.litellm.traces", + ) + ) + + +def test_a_misconfigured_logger_fails_at_startup_not_at_first_flush(monkeypatch): + for name in ("WORKSPACE_URL", "SERVER_ENDPOINT", "CLIENT_ID", "CLIENT_SECRET", "TABLE_NAME"): + monkeypatch.delenv(f"ZEROBUS_{name}", raising=False) + monkeypatch.setattr(litellm, "zerobus_params", None) + + with pytest.raises(ValueError, match="ZEROBUS_"): + ZerobusLogger() + + +def test_litellm_zerobus_params_configure_the_logger(monkeypatch): + monkeypatch.setattr( + litellm, + "zerobus_params", + { + "workspace_url": WORKSPACE_URL, + "server_endpoint": SERVER_ENDPOINT, + "client_id": "sp-id", + "client_secret": "sp-secret", + "table_name": "main.litellm.traces", + "batch_size": 7, + "flush_interval": 3, + }, + ) + + logger = ZerobusLogger() + + assert logger.batch_size == 7 + assert logger.flush_interval == 3 + assert logger.client.connection.table_name == "main.litellm.traces" + + +def test_the_client_is_kept_while_the_connection_is_unchanged_and_rebuilt_when_it_changes(monkeypatch): + """The client caches its token, so it must survive across flushes, yet a UI edit must take effect.""" + monkeypatch.setenv("ZEROBUS_WORKSPACE_URL", WORKSPACE_URL) + monkeypatch.setenv("ZEROBUS_SERVER_ENDPOINT", SERVER_ENDPOINT) + monkeypatch.setenv("ZEROBUS_CLIENT_ID", "sp-id") + monkeypatch.setenv("ZEROBUS_CLIENT_SECRET", "sp-secret") + monkeypatch.setenv("ZEROBUS_TABLE_NAME", "main.litellm.traces") + monkeypatch.setattr(litellm, "zerobus_params", None) + logger = ZerobusLogger() + + first = logger.client + unchanged = logger.client + monkeypatch.setenv("ZEROBUS_TABLE_NAME", "main.litellm.traces_v2") + rebuilt = logger.client + + assert unchanged is first + assert rebuilt is not first + assert rebuilt.connection.table_name == "main.litellm.traces_v2" diff --git a/tests/test_litellm/integrations/zerobus/test_row.py b/tests/test_litellm/integrations/zerobus/test_row.py new file mode 100644 index 00000000000..5a84009d66e --- /dev/null +++ b/tests/test_litellm/integrations/zerobus/test_row.py @@ -0,0 +1,137 @@ +import json + +from litellm.integrations.zerobus.row import TRACE_TABLE_COLUMNS, create_table_sql, trace_row + + +def _payload() -> dict: + return { + "id": "chatcmpl-1", + "trace_id": "trace-1", + "session_id": "session-1", + "litellm_call_id": "call-1", + "call_type": "acompletion", + "status": "success", + "model": "gpt-4o", + "model_group": "gpt-4o-group", + "custom_llm_provider": "openai", + "api_base": "https://api.openai.com", + "stream": False, + "cache_hit": None, + "startTime": 1_700_000_000.25, + "endTime": 1_700_000_001.5, + "completionStartTime": 1_700_000_000.75, + "response_time": 1.25, + "prompt_tokens": 10, + "completion_tokens": 5, + "total_tokens": 15, + "response_cost": 0.0015, + "saved_cache_cost": 0.0, + "end_user": "end-user-1", + "requester_ip_address": "10.0.0.1", + "user_agent": "curl/8", + "request_tags": ["prod"], + "messages": [{"role": "user", "content": "hi"}], + "response": {"choices": [{"message": {"role": "assistant", "content": "hello"}}]}, + "error_str": None, + "error_information": None, + "metadata": { + "user_api_key_hash": "hash-1", + "user_api_key_alias": "alias-1", + "user_api_key_team_id": "team-1", + "user_api_key_team_alias": "team-alias-1", + "user_api_key_user_id": "user-1", + "user_api_key_org_id": "org-1", + }, + "model_parameters": {"temperature": 0.2}, + "hidden_params": {"response_cost": 0.0015}, + "guardrail_information": None, + "cost_breakdown": {"input_cost": 0.001, "output_cost": 0.0005}, + } + + +def test_every_row_has_exactly_the_documented_columns(): + """Zerobus rejects a record naming a column the table lacks, so the row and the DDL must agree.""" + assert tuple(trace_row(_payload())) == tuple(TRACE_TABLE_COLUMNS) + assert tuple(trace_row({})) == tuple(TRACE_TABLE_COLUMNS) + + +def test_scalars_land_in_their_columns(): + row = trace_row(_payload()) + + assert row["id"] == "chatcmpl-1" + assert row["trace_id"] == "trace-1" + assert row["status"] == "success" + assert row["model"] == "gpt-4o" + assert row["stream"] is False + assert row["prompt_tokens"] == 10 + assert row["total_tokens"] == 15 + assert row["response_cost"] == 0.0015 + assert row["end_user"] == "end-user-1" + + +def test_key_and_team_identity_is_lifted_out_of_metadata(): + """Filtering spend by team or key is the main query, so those live in their own columns.""" + row = trace_row(_payload()) + + assert row["api_key_hash"] == "hash-1" + assert row["api_key_alias"] == "alias-1" + assert row["team_id"] == "team-1" + assert row["team_alias"] == "team-alias-1" + assert row["user_id"] == "user-1" + assert row["org_id"] == "org-1" + + +def test_timestamps_become_epoch_microseconds(): + row = trace_row(_payload()) + + assert row["start_time"] == 1_700_000_000_250_000 + assert row["end_time"] == 1_700_000_001_500_000 + assert row["completion_start_time"] == 1_700_000_000_750_000 + + +def test_a_zero_timestamp_is_null_rather_than_1970(): + """LiteLLM leaves completionStartTime at 0 when there is no first token, which is not a real time.""" + row = trace_row({**_payload(), "completionStartTime": 0}) + + assert row["completion_start_time"] is None + + +def test_nested_fields_are_json_text_for_the_variant_columns(): + row = trace_row(_payload()) + + assert json.loads(str(row["messages"])) == [{"role": "user", "content": "hi"}] + assert json.loads(str(row["metadata"]))["user_api_key_team_id"] == "team-1" + assert json.loads(str(row["request_tags"])) == ["prod"] + assert json.loads(str(row["cost_breakdown"])) == {"input_cost": 0.001, "output_cost": 0.0005} + + +def test_missing_and_null_fields_are_null(): + row = trace_row({**_payload(), "messages": None, "guardrail_information": None}) + + assert row["messages"] is None + assert row["guardrail_information"] is None + assert row["error_str"] is None + assert row["cache_hit"] is None + + +def test_a_wrongly_typed_field_is_null_instead_of_a_rejected_record(): + """One odd payload must not poison the whole batch: the table type wins.""" + row = trace_row({**_payload(), "prompt_tokens": "ten", "stream": "yes", "startTime": "now"}) + + assert row["prompt_tokens"] is None + assert row["stream"] is None + assert row["start_time"] is None + + +def test_the_row_is_json_serializable(): + json.dumps(dict(trace_row(_payload()))) + + +def test_create_table_sql_declares_every_column_with_its_type(): + sql = create_table_sql("main.litellm.traces") + + assert sql.startswith("CREATE TABLE main.litellm.traces (") + assert " start_time TIMESTAMP," in sql + assert " messages VARIANT," in sql + assert " cost_breakdown VARIANT\n);" in sql + assert sql.count(",") == len(TRACE_TABLE_COLUMNS) - 1 diff --git a/tests/test_litellm/proxy/test_zerobus_dashboard_config.py b/tests/test_litellm/proxy/test_zerobus_dashboard_config.py new file mode 100644 index 00000000000..e64265ad8c3 --- /dev/null +++ b/tests/test_litellm/proxy/test_zerobus_dashboard_config.py @@ -0,0 +1,40 @@ +import json +from pathlib import Path + +import litellm +from litellm.integrations.custom_logger import CustomLogger + + +def _dashboard_configs() -> tuple[dict, ...]: + path = Path(litellm.__file__).parent / "integrations" / "callback_configs.json" + return tuple(json.loads(path.read_text())) + + +def _zerobus_config() -> dict: + return next(config for config in _dashboard_configs() if config["id"] == "zerobus") + + +def test_zerobus_appears_in_the_dashboard_callback_dropdown(): + """The dropdown is served from callback_configs.json, so an entry only in the dashboard source is invisible.""" + entry = _zerobus_config() + + assert entry["displayName"] == "Databricks Zerobus" + assert entry["supports_key_team_logging"] is False + assert entry["dynamic_params"]["ZEROBUS_CLIENT_SECRET"]["type"] == "password" + assert all(field["required"] is True for field in entry["dynamic_params"].values()) + + +def test_the_dropdown_logo_asset_exists(): + """A logo the dashboard cannot resolve degrades silently to a letter tile.""" + logo = _zerobus_config()["logo"] + repo_root = Path(litellm.__file__).parent.parent + asset = repo_root / "ui" / "litellm-dashboard" / "public" / "assets" / "logos" / logo + + assert asset.is_file() + + +def test_the_dropdown_fields_are_the_env_vars_the_logger_reads(): + """Naming the fields as stored means the edit form prefills saved values instead of showing blanks.""" + fields = tuple(_zerobus_config()["dynamic_params"]) + + assert fields == tuple(CustomLogger.get_callback_env_vars("zerobus")) diff --git a/ui/litellm-dashboard/src/components/callback_info_helpers.tsx b/ui/litellm-dashboard/src/components/callback_info_helpers.tsx index 8e343e1a6e0..49ff8c8943c 100644 --- a/ui/litellm-dashboard/src/components/callback_info_helpers.tsx +++ b/ui/litellm-dashboard/src/components/callback_info_helpers.tsx @@ -10,6 +10,7 @@ import newrelicLogo from "../../public/assets/logos/newrelic.png"; import openmeterLogo from "../../public/assets/logos/openmeter.png"; import otelLogo from "../../public/assets/logos/otel.png"; import pointfiveLogo from "../../public/assets/logos/pointfive.png"; +import databricksLogo from "../../public/assets/logos/databricks.svg"; interface CallbackConfig { id: string; @@ -174,6 +175,20 @@ export const CALLBACK_CONFIGS: CallbackConfig[] = [ }, description: "PointFive Logging Integration", }, + { + id: "zerobus", + displayName: "Databricks Zerobus", + logo: databricksLogo.src, + supports_key_team_logging: false, + dynamic_params: { + ZEROBUS_WORKSPACE_URL: "text", + ZEROBUS_SERVER_ENDPOINT: "text", + ZEROBUS_CLIENT_ID: "text", + ZEROBUS_CLIENT_SECRET: "password", + ZEROBUS_TABLE_NAME: "text", + }, + description: "Databricks Zerobus Ingest Logging Integration", + }, { id: "s3", displayName: "S3",