feat(integrations): add Databricks Zerobus trace logging callback

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

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

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://<workspace-id>.zerobus.<region>.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",

litellm/integrations/zerobus/__init__.py

@@ -0,0 +1,5 @@
+"""Databricks Zerobus logging integration for LiteLLM."""
+
+from litellm.integrations.zerobus.logger import ZerobusLogger
+
+__all__ = ("ZerobusLogger",)

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)

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://<workspace_id>.zerobus.<region>.<cloud>``, 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://<workspace_id>.zerobus.<region>.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)

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"),
+        }
+    )

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,

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):

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):
     """

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

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

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"

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

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"))

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",