feat(logging): add ECS logging support via LITELLM_ECS_LOGS env var

Add ECSFormatter to litellm/_logging.py that emits logs conforming to
Elastic Common Schema v8.x. The existing JsonFormatter output does not
map to ECS field names (@timestamp, log.level, log.origin.*, etc.), so
ECS consumers (Elastic Stack, Datadog ECS mode, etc.) cannot ingest
litellm logs without a custom pipeline transform.

Changes:
- ECSFormatter class: produces @timestamp (UTC ISO-8601 with ms), nested
  log.{level,logger,origin.file.{name,line},origin.function}, message,
  service.name, ecs.version, and error.{type,message,stack_trace} on
  exceptions. Extra fields from logger(..., extra={}) pass through at the
  top level, excluding ECS reserved keys.
- LITELLM_ECS_LOGS=true env var: activates ECS formatting at startup.
  Takes precedence over JSON_LOGS since ECS is a superset of structured JSON.
  LITELLM_SERVICE_NAME can override the service.name field (default: litellm).
- _turn_on_ecs(): public function to switch all litellm loggers to ECS
  format at runtime, safe to call multiple times.
- litellm/sitecustomize.py: a drop-in file for the Python environment's
  site-packages that applies ECS logging before any application code runs
  when LITELLM_ECS_LOGS=true is set. Zero code changes required.
- 7 new regression tests covering field structure, timestamp format,
  lowercase level, error fields, extra passthrough, reserved key collision,
  and integration via _turn_on_ecs().

litellm/_logging.py

@@ -2,7 +2,7 @@ import ast
 import logging
 import os
 import sys
-from datetime import datetime
+from datetime import datetime, timezone
 from logging import Formatter
 from typing import Any, Dict, Optional
 
@@ -82,6 +82,7 @@ _secret_filter = SecretRedactionFilter()
 
 
 json_logs = bool(os.getenv("JSON_LOGS", False))
+ecs_logs = os.getenv("LITELLM_ECS_LOGS", "").lower() == "true"
 # Create a handler for the logger (you may need to adapt this based on your needs)
 log_level = os.getenv("LITELLM_LOG", "DEBUG")
 numeric_level: str = getattr(logging, log_level.upper())
@@ -196,6 +197,72 @@ class JsonFormatter(Formatter):
         return safe_dumps(json_record)
 
 
+_ECS_RESERVED_KEYS = frozenset(
+    {"@timestamp", "log", "message", "service", "ecs", "error"}
+)
+
+
+class ECSFormatter(Formatter):
+    """Formats log records according to Elastic Common Schema (ECS) v8.x.
+
+    Enables structured log ingestion into the Elastic Stack and other ECS-aware
+    platforms. Activate via the LITELLM_ECS_LOGS=true environment variable or
+    call _turn_on_ecs() at application startup.
+
+    Reference: https://www.elastic.co/guide/en/ecs/current/index.html
+    """
+
+    ECS_VERSION = "8.11.0"
+
+    def __init__(
+        self,
+        service_name: str = os.getenv("LITELLM_SERVICE_NAME", "litellm"),
+    ):
+        super().__init__()
+        self._service_name = service_name
+
+    def formatTime(
+        self, record: logging.LogRecord, datefmt: Optional[str] = None
+    ) -> str:
+        dt = datetime.fromtimestamp(record.created, tz=timezone.utc)
+        return dt.strftime("%Y-%m-%dT%H:%M:%S.") + f"{dt.microsecond // 1000:03d}Z"
+
+    def format(self, record: logging.LogRecord) -> str:
+        message_str = record.getMessage()
+
+        ecs_record: Dict[str, Any] = {
+            "@timestamp": self.formatTime(record),
+            "log": {
+                "level": record.levelname.lower(),
+                "logger": record.name,
+                "origin": {
+                    "file": {
+                        "name": record.filename,
+                        "line": record.lineno,
+                    },
+                    "function": record.funcName,
+                },
+            },
+            "message": message_str,
+            "service": {"name": self._service_name},
+            "ecs": {"version": self.ECS_VERSION},
+        }
+
+        if record.exc_info and record.exc_info[1] is not None:
+            exc_type, exc_value, _ = record.exc_info
+            ecs_record["error"] = {
+                "type": exc_type.__name__ if exc_type else None,
+                "message": str(exc_value),
+                "stack_trace": record.exc_text or self.formatException(record.exc_info),
+            }
+
+        for key, value in record.__dict__.items():
+            if key not in _STANDARD_RECORD_ATTRS and key not in _ECS_RESERVED_KEYS:
+                ecs_record[key] = value
+
+        return safe_dumps(ecs_record)
+
+
 # Function to set up exception handlers for JSON logging
 def _setup_json_exception_handlers(formatter):
     # Create a handler with JSON formatting for exceptions
@@ -244,8 +311,12 @@ def _setup_json_exception_handlers(formatter):
         pass
 
 
-# Create a formatter and set it for the handler
-if json_logs:
+# Create a formatter and set it for the handler.
+# LITELLM_ECS_LOGS takes precedence over JSON_LOGS since ECS is a superset of structured JSON.
+if ecs_logs:
+    handler.setFormatter(ECSFormatter())
+    _setup_json_exception_handlers(ECSFormatter())
+elif json_logs:
     handler.setFormatter(JsonFormatter())
     _setup_json_exception_handlers(JsonFormatter())
 else:
@@ -385,18 +456,23 @@ def _get_uvicorn_json_log_config():
 
 
 def _turn_on_json():
-    """
-    Turn on JSON logging
-
-    - Adds a JSON formatter to all loggers
-    """
     handler = logging.StreamHandler()
     handler.setFormatter(JsonFormatter())
     _initialize_loggers_with_handler(handler)
-    # Set up exception handlers
     _setup_json_exception_handlers(JsonFormatter())
 
 
+def _turn_on_ecs():
+    """Switch all litellm loggers to ECS-formatted JSON output.
+
+    Idempotent; safe to call multiple times or from sitecustomize.py.
+    """
+    handler = logging.StreamHandler()
+    handler.setFormatter(ECSFormatter())
+    _initialize_loggers_with_handler(handler)
+    _setup_json_exception_handlers(ECSFormatter())
+
+
 def _turn_on_debug():
     verbose_logger.setLevel(level=logging.DEBUG)  # set package log to debug
     verbose_router_logger.setLevel(level=logging.DEBUG)  # set router logs to debug

litellm/sitecustomize.py

@@ -0,0 +1,34 @@
+"""
+ECS logging early-startup hook for litellm.
+
+Python executes sitecustomize.py from the active site-packages directory
+before any user code, making it the right place to configure logging format
+before the first log record is emitted.
+
+To enable ECS-compliant log output without touching application code:
+
+    1. Find your environment's site-packages path:
+           python -c "import site; print(site.getsitepackages()[0])"
+
+    2. If no sitecustomize.py exists there yet, copy this file:
+           cp litellm/sitecustomize.py <site-packages>/sitecustomize.py
+
+       If one already exists, append this import to it:
+           echo "import litellm.sitecustomize" >> <site-packages>/sitecustomize.py
+
+    3. Set the environment variable before starting your process:
+           export LITELLM_ECS_LOGS=true
+
+Alternatively, call litellm._logging._turn_on_ecs() directly at the top of
+your application's entrypoint if you prefer not to touch site-packages.
+"""
+
+import os
+
+if os.environ.get("LITELLM_ECS_LOGS", "").lower() == "true":
+    try:
+        from litellm._logging import _turn_on_ecs
+
+        _turn_on_ecs()
+    except Exception:
+        pass

tests/test_litellm/test_logging.py

@@ -1,6 +1,7 @@
 import asyncio
 import json
 import os
+import re
 import sys
 from typing import List
 
@@ -15,8 +16,10 @@ import sys
 import litellm
 from litellm._logging import (
     ALL_LOGGERS,
+    ECSFormatter,
     JsonFormatter,
     _initialize_loggers_with_handler,
+    _turn_on_ecs,
     _turn_on_json,
     verbose_logger,
     verbose_proxy_logger,
@@ -328,3 +331,151 @@ async def test_cache_hit_includes_custom_llm_provider():
         # Clean up
         litellm.callbacks = original_callbacks
         litellm.cache = None
+
+
+# ---------------------------------------------------------------------------
+# ECS formatter tests
+# ---------------------------------------------------------------------------
+
+
+def test_ecs_formatter_required_fields():
+    formatter = ECSFormatter()
+    record = logging.LogRecord(
+        name="LiteLLM",
+        level=logging.INFO,
+        pathname="proxy_server.py",
+        lineno=42,
+        msg="test message",
+        args=(),
+        exc_info=None,
+    )
+    obj = json.loads(formatter.format(record))
+
+    assert obj["message"] == "test message"
+    assert "@timestamp" in obj
+    assert obj["log"]["level"] == "info"
+    assert obj["log"]["logger"] == "LiteLLM"
+    assert obj["log"]["origin"]["file"]["name"] == "proxy_server.py"
+    assert obj["log"]["origin"]["file"]["line"] == 42
+    assert obj["service"]["name"] == "litellm"
+    assert obj["ecs"]["version"] == "8.11.0"
+
+
+def test_ecs_formatter_timestamp_is_utc_iso8601_with_ms():
+    formatter = ECSFormatter()
+    record = logging.LogRecord(
+        name="LiteLLM",
+        level=logging.DEBUG,
+        pathname="",
+        lineno=0,
+        msg="ts test",
+        args=(),
+        exc_info=None,
+    )
+    obj = json.loads(formatter.format(record))
+    assert re.match(
+        r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$", obj["@timestamp"]
+    ), f"Non-ECS timestamp: {obj['@timestamp']!r}"
+
+
+def test_ecs_formatter_log_level_is_lowercase():
+    formatter = ECSFormatter()
+    for level, expected in [
+        (logging.DEBUG, "debug"),
+        (logging.INFO, "info"),
+        (logging.WARNING, "warning"),
+        (logging.ERROR, "error"),
+        (logging.CRITICAL, "critical"),
+    ]:
+        record = logging.LogRecord(
+            name="LiteLLM",
+            level=level,
+            pathname="",
+            lineno=0,
+            msg="test",
+            args=(),
+            exc_info=None,
+        )
+        obj = json.loads(formatter.format(record))
+        assert obj["log"]["level"] == expected
+
+
+def test_ecs_formatter_error_fields_on_exception():
+    formatter = ECSFormatter()
+    try:
+        raise ValueError("something broke")
+    except ValueError:
+        exc_info = sys.exc_info()
+
+    record = logging.LogRecord(
+        name="LiteLLM",
+        level=logging.ERROR,
+        pathname="",
+        lineno=0,
+        msg="error occurred",
+        args=(),
+        exc_info=exc_info,
+    )
+    record.exc_text = formatter.formatException(exc_info)
+    obj = json.loads(formatter.format(record))
+
+    assert "error" in obj
+    assert obj["error"]["type"] == "ValueError"
+    assert obj["error"]["message"] == "something broke"
+    assert "stack_trace" in obj["error"]
+    assert "ValueError" in obj["error"]["stack_trace"]
+
+
+def test_ecs_formatter_extra_fields_passthrough():
+    formatter = ECSFormatter()
+    record = logging.LogRecord(
+        name="LiteLLM",
+        level=logging.DEBUG,
+        pathname="",
+        lineno=0,
+        msg="request received",
+        args=(),
+        exc_info=None,
+    )
+    record.api_base = "https://api.openai.com"
+    record.model = "gpt-4"
+    obj = json.loads(formatter.format(record))
+
+    assert obj["api_base"] == "https://api.openai.com"
+    assert obj["model"] == "gpt-4"
+
+
+def test_ecs_formatter_no_ecs_reserved_key_collision():
+    formatter = ECSFormatter()
+    record = logging.LogRecord(
+        name="LiteLLM",
+        level=logging.INFO,
+        pathname="",
+        lineno=0,
+        msg="test",
+        args=(),
+        exc_info=None,
+    )
+    # Attempt to inject via extra - should not overwrite ECS structure
+    record.message = "injected"
+    obj = json.loads(formatter.format(record))
+    assert obj["message"] == "test"
+
+
+def test_ecs_mode_emits_ecs_fields(capfd):
+    _turn_on_ecs()
+    for lg in (verbose_logger, verbose_router_logger, verbose_proxy_logger):
+        lg.setLevel(logging.INFO)
+
+    verbose_logger.info("ecs integration test")
+
+    _, err = capfd.readouterr()
+    lines = [line for line in err.splitlines() if line.strip()]
+    assert lines, "Expected at least one log line"
+
+    obj = json.loads(lines[-1])
+    assert "@timestamp" in obj
+    assert obj["log"]["level"] == "info"
+    assert obj["message"] == "ecs integration test"
+    assert obj["ecs"]["version"] == "8.11.0"
+    assert obj["service"]["name"] == "litellm"