diff --git a/docs/my-website/docs/observability/levo_integration.md b/docs/my-website/docs/observability/levo_integration.md new file mode 100644 index 0000000000..3e46cf6b92 --- /dev/null +++ b/docs/my-website/docs/observability/levo_integration.md @@ -0,0 +1,162 @@ +--- +sidebar_label: Levo AI +--- + +import Image from '@theme/IdealImage'; +import Tabs from '@theme/Tabs'; +import TabItem from '@theme/TabItem'; + +# Levo AI + +
+
+ +
+
+ +
+
+ +[Levo](https://levo.ai/) is an AI observability and compliance platform that provides comprehensive monitoring, analysis, and compliance tracking for LLM applications. + +## Quick Start + +Send all your LLM requests and responses to Levo for monitoring and analysis using LiteLLM's built-in Levo integration. + +### What You'll Get + +- **Complete visibility** into all LLM API calls across all providers +- **Request and response data** including prompts, completions, and metadata +- **Usage and cost tracking** with token counts and cost breakdowns +- **Error monitoring** and performance metrics +- **Compliance tracking** for audit and governance + +### Setup Steps + +**1. Install OpenTelemetry dependencies:** + +```bash +pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp-proto-http opentelemetry-exporter-otlp-proto-grpc +``` + +**2. Enable Levo callback in your LiteLLM config:** + +Add to your `litellm_config.yaml`: + +```yaml +litellm_settings: + callbacks: ["levo"] +``` + +**3. Configure environment variables:** + +[Contact Levo support](mailto:support@levo.ai) to get your collector endpoint URL, API key, organization ID, and workspace ID. + +Set these required environment variables: + +```bash +export LEVOAI_API_KEY="" +export LEVOAI_ORG_ID="" +export LEVOAI_WORKSPACE_ID="" +export LEVOAI_COLLECTOR_URL="" +``` + +**Note:** The collector URL should be the full endpoint URL provided by Levo support. It will be used exactly as provided. + +**4. Start LiteLLM:** + +```bash +litellm --config config.yaml +``` + +**5. Make requests - they'll automatically be sent to Levo!** + +```bash +curl --location 'http://0.0.0.0:4000/chat/completions' \ + --header 'Content-Type: application/json' \ + --data '{ + "model": "gpt-3.5-turbo", + "messages": [ + { + "role": "user", + "content": "Hello, this is a test message" + } + ] + }' +``` + +## What Data is Captured + +| Feature | Details | +|---------|---------| +| **What is logged** | OpenTelemetry Trace Data (OTLP format) | +| **Events** | Success + Failure | +| **Format** | OTLP (OpenTelemetry Protocol) | +| **Headers** | Automatically includes `Authorization: Bearer {LEVOAI_API_KEY}`, `x-levo-organization-id`, and `x-levo-workspace-id` | + +## Configuration Reference + +### Required Environment Variables + +| Variable | Description | Example | +|----------|-------------|---------| +| `LEVOAI_API_KEY` | Your Levo API key | `levo_abc123...` | +| `LEVOAI_ORG_ID` | Your Levo organization ID | `org-123456` | +| `LEVOAI_WORKSPACE_ID` | Your Levo workspace ID | `workspace-789` | +| `LEVOAI_COLLECTOR_URL` | Full collector endpoint URL from Levo support | `https://collector.levo.ai/v1/traces` | + +### Optional Environment Variables + +| Variable | Description | Default | +|----------|-------------|---------| +| `LEVOAI_ENV_NAME` | Environment name for tagging traces | `None` | + +**Note:** The collector URL is used exactly as provided by Levo support. No path manipulation is performed. + +## Troubleshooting + +### Not seeing traces in Levo? + +1. **Verify Levo callback is enabled**: Check LiteLLM startup logs for `initializing callbacks=['levo']` + +2. **Check required environment variables**: Ensure all required variables are set: + ```bash + echo $LEVOAI_API_KEY + echo $LEVOAI_ORG_ID + echo $LEVOAI_WORKSPACE_ID + echo $LEVOAI_COLLECTOR_URL + ``` + +3. **Verify collector connectivity**: Test if your collector is reachable: + ```bash + curl /health + ``` + +4. **Check for initialization errors**: Look for errors in LiteLLM startup logs. Common issues: + - Missing OpenTelemetry packages: Install with `pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp-proto-http opentelemetry-exporter-otlp-proto-grpc` + - Missing required environment variables: All four required variables must be set + - Invalid collector URL: Ensure the URL is correct and reachable + +5. **Enable debug logging**: + ```bash + export LITELLM_LOG="DEBUG" + ``` + +6. **Wait for async export**: OTLP sends traces asynchronously. Wait 10-15 seconds after making requests before checking Levo. + +### Common Errors + +**Error: "LEVOAI_COLLECTOR_URL environment variable is required"** +- Solution: Set the `LEVOAI_COLLECTOR_URL` environment variable with your collector endpoint URL from Levo support. + +**Error: "No module named 'opentelemetry'"** +- Solution: Install OpenTelemetry packages: `pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp-proto-http opentelemetry-exporter-otlp-proto-grpc` + +## Additional Resources + +- [Levo Documentation](https://docs.levo.ai) +- [OpenTelemetry Specification](https://opentelemetry.io/docs/specs/otel/) + +## Need Help? + +For issues or questions about the Levo integration with LiteLLM, please [contact Levo support](mailto:support@levo.ai) or open an issue on the [LiteLLM GitHub repository](https://github.com/BerriAI/litellm/issues). diff --git a/docs/my-website/docs/observability/opentelemetry_integration.md b/docs/my-website/docs/observability/opentelemetry_integration.md index aa59d0aa3f..b6eff23162 100644 --- a/docs/my-website/docs/observability/opentelemetry_integration.md +++ b/docs/my-website/docs/observability/opentelemetry_integration.md @@ -4,7 +4,7 @@ import TabItem from '@theme/TabItem'; # OpenTelemetry - Tracing LLMs with any observability tool -OpenTelemetry is a CNCF standard for observability. It connects to any observability tool, such as Jaeger, Zipkin, Datadog, New Relic, Traceloop and others. +OpenTelemetry is a CNCF standard for observability. It connects to any observability tool, such as Jaeger, Zipkin, Datadog, New Relic, Traceloop, Levo AI and others. diff --git a/docs/my-website/img/levo_logo.png b/docs/my-website/img/levo_logo.png new file mode 100644 index 0000000000..fdb72470b2 Binary files /dev/null and b/docs/my-website/img/levo_logo.png differ diff --git a/docs/my-website/img/levo_logo_dark.png b/docs/my-website/img/levo_logo_dark.png new file mode 100644 index 0000000000..70da632ee9 Binary files /dev/null and b/docs/my-website/img/levo_logo_dark.png differ diff --git a/docs/my-website/src/css/custom.css b/docs/my-website/src/css/custom.css index 2bc6a4cfde..9fa4443afc 100644 --- a/docs/my-website/src/css/custom.css +++ b/docs/my-website/src/css/custom.css @@ -28,3 +28,34 @@ --ifm-color-primary-lightest: #4fddbf; --docusaurus-highlighted-code-line-bg: rgba(0, 0, 0, 0.3); } + +/* Levo logo sizing and theme switching */ +.levo-logo-container { + position: relative; +} + +.levo-logo-container img, +.levo-logo-container picture, +.levo-logo-container .ideal-image { + max-width: 200px !important; + width: 200px !important; + height: auto !important; +} + +/* Show light logo by default, hide dark logo */ +.levo-logo-dark { + display: none !important; +} + +.levo-logo-light { + display: block !important; +} + +/* In dark mode, hide light logo and show dark logo */ +[data-theme='dark'] .levo-logo-light { + display: none !important; +} + +[data-theme='dark'] .levo-logo-dark { + display: block !important; +} diff --git a/litellm/__init__.py b/litellm/__init__.py index 207ece224f..4f8c57ef44 100644 --- a/litellm/__init__.py +++ b/litellm/__init__.py @@ -136,6 +136,7 @@ _custom_logger_compatible_callbacks_literal = Literal[ "gitlab", "cloudzero", "posthog", + "levo", ] cold_storage_custom_logger: Optional[_custom_logger_compatible_callbacks_literal] = None logged_real_time_event_types: Optional[Union[List[str], Literal["*"]]] = None diff --git a/litellm/integrations/levo/README.md b/litellm/integrations/levo/README.md new file mode 100644 index 0000000000..cb18b1dbfb --- /dev/null +++ b/litellm/integrations/levo/README.md @@ -0,0 +1,125 @@ +# Levo AI Integration + +This integration enables sending LLM observability data to Levo AI using OpenTelemetry (OTLP) protocol. + +## Overview + +The Levo integration extends LiteLLM's OpenTelemetry support to automatically send traces to Levo's collector endpoint with proper authentication and routing headers. + +## Features + +- **Automatic OTLP Export**: Sends OpenTelemetry traces to Levo collector +- **Levo-Specific Headers**: Automatically includes `x-levo-organization-id` and `x-levo-workspace-id` for routing +- **Simple Configuration**: Just use `callbacks: ["levo"]` in your LiteLLM config +- **Environment-Based Setup**: Configure via environment variables + +## Quick Start + +### 1. Install Dependencies + +```bash +pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp-proto-http opentelemetry-exporter-otlp-proto-grpc +``` + +### 2. Configure LiteLLM + +Add to your `litellm_config.yaml`: + +```yaml +litellm_settings: + callbacks: ["levo"] +``` + +### 3. Set Environment Variables + +```bash +export LEVOAI_API_KEY="" +export LEVOAI_ORG_ID="" +export LEVOAI_WORKSPACE_ID="" +export LEVOAI_COLLECTOR_URL="" +``` + +### 4. Start LiteLLM + +```bash +litellm --config config.yaml +``` + +All LLM requests will now automatically be sent to Levo! + +## Configuration + +### Required Environment Variables + +| Variable | Description | +|----------|-------------| +| `LEVOAI_API_KEY` | Your Levo API key for authentication | +| `LEVOAI_ORG_ID` | Your Levo organization ID for routing | +| `LEVOAI_WORKSPACE_ID` | Your Levo workspace ID for routing | +| `LEVOAI_COLLECTOR_URL` | Full collector endpoint URL from Levo support | + +### Optional Environment Variables + +| Variable | Description | Default | +|----------|-------------|---------| +| `LEVOAI_ENV_NAME` | Environment name for tagging traces | `None` | + +**Important**: The `LEVOAI_COLLECTOR_URL` is used exactly as provided. No path manipulation is performed. + +## How It Works + +1. **LevoLogger** extends LiteLLM's `OpenTelemetry` class +2. **Configuration** is read from environment variables via `get_levo_config()` +3. **OTLP Headers** are automatically set: + - `Authorization: Bearer {LEVOAI_API_KEY}` + - `x-levo-organization-id: {LEVOAI_ORG_ID}` + - `x-levo-workspace-id: {LEVOAI_WORKSPACE_ID}` +4. **Traces** are sent to the collector endpoint in OTLP format + +## Code Structure + +``` +litellm/integrations/levo/ +├── __init__.py # Exports LevoLogger +├── levo.py # LevoLogger implementation +└── README.md # This file +``` + +### Key Classes + +- **LevoLogger**: Extends `OpenTelemetry`, handles Levo-specific configuration +- **LevoConfig**: Pydantic model for Levo configuration (defined in `levo.py`) + +## Testing + +See the test files in `tests/test_litellm/integrations/levo/`: +- `test_levo.py`: Unit tests for configuration +- `test_levo_integration.py`: Integration tests for callback registration + +## Error Handling + +The integration validates all required environment variables at initialization: +- Missing `LEVOAI_API_KEY`: Raises `ValueError` with clear message +- Missing `LEVOAI_ORG_ID`: Raises `ValueError` with clear message +- Missing `LEVOAI_WORKSPACE_ID`: Raises `ValueError` with clear message +- Missing `LEVOAI_COLLECTOR_URL`: Raises `ValueError` with clear message + +## Integration with LiteLLM + +The Levo callback is registered in: +- `litellm/litellm_core_utils/custom_logger_registry.py`: Maps `"levo"` to `LevoLogger` +- `litellm/litellm_core_utils/litellm_logging.py`: Instantiates `LevoLogger` when `callbacks: ["levo"]` is used +- `litellm/__init__.py`: Added to `_custom_logger_compatible_callbacks_literal` + +## Documentation + +For detailed documentation, see: +- [LiteLLM Levo Integration Docs](../../../../docs/my-website/docs/observability/levo_integration.md) +- [Levo Documentation](https://docs.levo.ai) + +## Support + +For issues or questions: +- LiteLLM Issues: https://github.com/BerriAI/litellm/issues +- Levo Support: support@levo.ai + diff --git a/litellm/integrations/levo/__init__.py b/litellm/integrations/levo/__init__.py new file mode 100644 index 0000000000..7f4f84437d --- /dev/null +++ b/litellm/integrations/levo/__init__.py @@ -0,0 +1,3 @@ +from litellm.integrations.levo.levo import LevoLogger + +__all__ = ["LevoLogger"] diff --git a/litellm/integrations/levo/levo.py b/litellm/integrations/levo/levo.py new file mode 100644 index 0000000000..562f2fd906 --- /dev/null +++ b/litellm/integrations/levo/levo.py @@ -0,0 +1,117 @@ +import os +from typing import TYPE_CHECKING, Any, Optional, Union + +from litellm.integrations.opentelemetry import OpenTelemetry + +if TYPE_CHECKING: + from opentelemetry.trace import Span as _Span + + from litellm.integrations.opentelemetry import OpenTelemetryConfig as _OpenTelemetryConfig + from litellm.types.integrations.arize import Protocol as _Protocol + + Protocol = _Protocol + OpenTelemetryConfig = _OpenTelemetryConfig + Span = Union[_Span, Any] +else: + Protocol = Any + OpenTelemetryConfig = Any + Span = Any + + +class LevoConfig: + """Configuration for Levo OTLP integration.""" + + def __init__( + self, + otlp_auth_headers: Optional[str], + protocol: Protocol, + endpoint: str, + ): + self.otlp_auth_headers = otlp_auth_headers + self.protocol = protocol + self.endpoint = endpoint + + +class LevoLogger(OpenTelemetry): + """Levo Logger that extends OpenTelemetry for OTLP integration.""" + + @staticmethod + def get_levo_config() -> LevoConfig: + """ + Retrieves the Levo configuration based on environment variables. + + Returns: + LevoConfig: Configuration object containing Levo OTLP settings. + + Raises: + ValueError: If required environment variables are missing. + """ + # Required environment variables + api_key = os.environ.get("LEVOAI_API_KEY", None) + org_id = os.environ.get("LEVOAI_ORG_ID", None) + workspace_id = os.environ.get("LEVOAI_WORKSPACE_ID", None) + collector_url = os.environ.get("LEVOAI_COLLECTOR_URL", None) + + # Validate required env vars + if not api_key: + raise ValueError( + "LEVOAI_API_KEY environment variable is required for Levo integration." + ) + if not org_id: + raise ValueError( + "LEVOAI_ORG_ID environment variable is required for Levo integration." + ) + if not workspace_id: + raise ValueError( + "LEVOAI_WORKSPACE_ID environment variable is required for Levo integration." + ) + if not collector_url: + raise ValueError( + "LEVOAI_COLLECTOR_URL environment variable is required for Levo integration. " + "Please contact Levo support to get your collector URL." + ) + + # Use collector URL exactly as provided by the user + endpoint = collector_url + protocol: Protocol = "otlp_http" + + # Build OTLP headers string + # Format: Authorization=Bearer {api_key},x-levo-organization-id={org_id},x-levo-workspace-id={workspace_id} + headers_parts = [f"Authorization=Bearer {api_key}"] + headers_parts.append(f"x-levo-organization-id={org_id}") + headers_parts.append(f"x-levo-workspace-id={workspace_id}") + + otlp_auth_headers = ",".join(headers_parts) + + return LevoConfig( + otlp_auth_headers=otlp_auth_headers, + protocol=protocol, + endpoint=endpoint, + ) + + async def async_health_check(self): + """ + Health check for Levo integration. + + Returns: + dict: Health status with status and message/error_message keys. + """ + try: + config = self.get_levo_config() + + if not config.otlp_auth_headers: + return { + "status": "unhealthy", + "error_message": "LEVOAI_API_KEY environment variable not set", + } + + return { + "status": "healthy", + "message": "Levo credentials are configured properly", + } + except ValueError as e: + return { + "status": "unhealthy", + "error_message": str(e), + } + diff --git a/litellm/litellm_core_utils/custom_logger_registry.py b/litellm/litellm_core_utils/custom_logger_registry.py index fa2ff42e1d..47cbcb8aec 100644 --- a/litellm/litellm_core_utils/custom_logger_registry.py +++ b/litellm/litellm_core_utils/custom_logger_registry.py @@ -76,6 +76,7 @@ class CustomLoggerRegistry: "arize_phoenix": OpenTelemetry, "langtrace": OpenTelemetry, "weave_otel": OpenTelemetry, + "levo": OpenTelemetry, "mlflow": MlflowLogger, "langfuse": LangfusePromptManagement, "otel": OpenTelemetry, diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 5c3c56e412..cd32493556 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -3699,6 +3699,31 @@ def _init_custom_logger_compatible_class( # noqa: PLR0915 ) _in_memory_loggers.append(_arize_phoenix_otel_logger) return _arize_phoenix_otel_logger # type: ignore + elif logging_integration == "levo": + from litellm.integrations.levo.levo import LevoLogger + from litellm.integrations.opentelemetry import ( + OpenTelemetry, + OpenTelemetryConfig, + ) + + levo_config = LevoLogger.get_levo_config() + otel_config = OpenTelemetryConfig( + exporter=levo_config.protocol, + endpoint=levo_config.endpoint, + headers=levo_config.otlp_auth_headers, + ) + + # Check if LevoLogger instance already exists + for callback in _in_memory_loggers: + if ( + isinstance(callback, LevoLogger) + and callback.callback_name == "levo" + ): + return callback # type: ignore + + _levo_otel_logger = LevoLogger(config=otel_config, callback_name="levo") + _in_memory_loggers.append(_levo_otel_logger) + return _levo_otel_logger # type: ignore elif logging_integration == "otel": from litellm.integrations.opentelemetry import OpenTelemetry diff --git a/tests/test_litellm/integrations/levo/__init__.py b/tests/test_litellm/integrations/levo/__init__.py new file mode 100644 index 0000000000..1560e78b7b --- /dev/null +++ b/tests/test_litellm/integrations/levo/__init__.py @@ -0,0 +1 @@ +# Levo integration tests diff --git a/tests/test_litellm/integrations/levo/test_levo.py b/tests/test_litellm/integrations/levo/test_levo.py new file mode 100644 index 0000000000..5d042cbc06 --- /dev/null +++ b/tests/test_litellm/integrations/levo/test_levo.py @@ -0,0 +1,407 @@ +import unittest +from unittest.mock import patch + +import pytest + +from litellm.integrations.levo.levo import LevoConfig, LevoLogger +from litellm.integrations.opentelemetry import OpenTelemetryConfig + +# Try to import OpenTelemetry packages, skip tests if not available +try: + from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( + InMemorySpanExporter, + ) + from opentelemetry.sdk.trace.export import SimpleSpanProcessor + + OPENTELEMETRY_AVAILABLE = True +except ImportError: + OPENTELEMETRY_AVAILABLE = False + + +class TestLevoConfig(unittest.TestCase): + """Unit tests for LevoLogger configuration.""" + + @patch.dict( + "os.environ", + { + "LEVOAI_API_KEY": "test-api-key", + "LEVOAI_ORG_ID": "test-org-id", + "LEVOAI_WORKSPACE_ID": "test-workspace-id", + "LEVOAI_COLLECTOR_URL": "https://collector.levo.ai", + }, + ) + def test_get_levo_config_with_all_required_vars(self): + """Test get_levo_config() with all required environment variables.""" + config = LevoLogger.get_levo_config() + + # Verify headers include all three values + self.assertIn("Authorization=Bearer test-api-key", config.otlp_auth_headers) + self.assertIn("x-levo-organization-id=test-org-id", config.otlp_auth_headers) + self.assertIn("x-levo-workspace-id=test-workspace-id", config.otlp_auth_headers) + + # Verify endpoint uses provided collector URL exactly as-is + self.assertEqual(config.endpoint, "https://collector.levo.ai") + + # Verify protocol is otlp_http + self.assertEqual(config.protocol, "otlp_http") + + @patch.dict( + "os.environ", + { + "LEVOAI_API_KEY": "test-api-key", + "LEVOAI_ORG_ID": "test-org-id", + "LEVOAI_WORKSPACE_ID": "test-workspace-id", + "LEVOAI_COLLECTOR_URL": "https://custom.collector.com", + }, + ) + def test_get_levo_config_with_custom_collector_url(self): + """Test get_levo_config() with custom collector URL.""" + config = LevoLogger.get_levo_config() + + # Verify endpoint uses custom URL exactly as provided + self.assertEqual(config.endpoint, "https://custom.collector.com") + self.assertEqual(config.protocol, "otlp_http") + + @patch.dict("os.environ", {}, clear=True) + def test_get_levo_config_missing_api_key(self): + """Test get_levo_config() raises ValueError when LEVOAI_API_KEY is missing.""" + with pytest.raises(ValueError, match="LEVOAI_API_KEY"): + LevoLogger.get_levo_config() + + @patch.dict( + "os.environ", + { + "LEVOAI_API_KEY": "test-api-key", + }, + clear=True, + ) + def test_get_levo_config_missing_org_id(self): + """Test get_levo_config() raises ValueError when LEVOAI_ORG_ID is missing.""" + with pytest.raises(ValueError, match="LEVOAI_ORG_ID"): + LevoLogger.get_levo_config() + + @patch.dict( + "os.environ", + { + "LEVOAI_API_KEY": "test-api-key", + "LEVOAI_ORG_ID": "test-org-id", + }, + clear=True, + ) + def test_get_levo_config_missing_workspace_id(self): + """Test get_levo_config() raises ValueError when LEVOAI_WORKSPACE_ID is missing.""" + with pytest.raises(ValueError, match="LEVOAI_WORKSPACE_ID"): + LevoLogger.get_levo_config() + + @patch.dict( + "os.environ", + { + "LEVOAI_API_KEY": "test-api-key", + "LEVOAI_ORG_ID": "test-org-id", + "LEVOAI_WORKSPACE_ID": "test-workspace-id", + }, + clear=True, + ) + def test_get_levo_config_missing_collector_url(self): + """Test get_levo_config() raises ValueError when LEVOAI_COLLECTOR_URL is missing.""" + with pytest.raises(ValueError, match="LEVOAI_COLLECTOR_URL"): + LevoLogger.get_levo_config() + + @patch.dict( + "os.environ", + { + "LEVOAI_API_KEY": "test-api-key", + "LEVOAI_ORG_ID": "test-org-id", + "LEVOAI_WORKSPACE_ID": "test-workspace-id", + "LEVOAI_COLLECTOR_URL": "http://localhost:4318", + }, + ) + def test_get_levo_config_with_http_endpoint(self): + """Test get_levo_config() with HTTP endpoint.""" + config = LevoLogger.get_levo_config() + + # Should use HTTP endpoint exactly as provided + self.assertEqual(config.endpoint, "http://localhost:4318") + self.assertEqual(config.protocol, "otlp_http") + + @patch.dict( + "os.environ", + { + "LEVOAI_API_KEY": "test-api-key", + "LEVOAI_ORG_ID": "test-org-id", + "LEVOAI_WORKSPACE_ID": "test-workspace-id", + "LEVOAI_COLLECTOR_URL": "https://collector.levo.ai", + }, + ) + def test_levo_config_headers_format(self): + """Test that OTLP headers are formatted correctly.""" + config = LevoLogger.get_levo_config() + + # Verify headers contain all required parts + self.assertIn("Authorization=Bearer test-api-key", config.otlp_auth_headers) + self.assertIn("x-levo-organization-id=test-org-id", config.otlp_auth_headers) + self.assertIn("x-levo-workspace-id=test-workspace-id", config.otlp_auth_headers) + + # Verify headers are comma-separated + header_parts = config.otlp_auth_headers.split(",") + self.assertEqual(len(header_parts), 3) + + +class TestLevoIntegration(unittest.TestCase): + """Integration tests for LevoLogger.""" + + @patch.dict( + "os.environ", + { + "LEVOAI_API_KEY": "test-api-key", + "LEVOAI_ORG_ID": "test-org-id", + "LEVOAI_WORKSPACE_ID": "test-workspace-id", + "LEVOAI_COLLECTOR_URL": "https://collector.levo.ai", + }, + ) + @pytest.mark.skipif( + not OPENTELEMETRY_AVAILABLE, reason="OpenTelemetry packages not installed" + ) + @patch( + "litellm.integrations.opentelemetry.OpenTelemetry._init_otel_logger_on_litellm_proxy" + ) + def test_levo_logger_instantiation(self, mock_init_proxy): + """Test that LevoLogger can be instantiated with proper config.""" + # Mock the proxy initialization to avoid importing proxy code + mock_init_proxy.return_value = None + + config = LevoLogger.get_levo_config() + otel_config = OpenTelemetryConfig( + exporter=config.protocol, + endpoint=config.endpoint, + headers=config.otlp_auth_headers, + ) + + # Create a tracer provider with in-memory exporter to avoid requiring OTLP packages + tracer_provider = TracerProvider() + tracer_provider.add_span_processor(SimpleSpanProcessor(InMemorySpanExporter())) + + # Create LevoLogger instance with mocked tracer provider + levo_logger = LevoLogger( + config=otel_config, callback_name="levo", tracer_provider=tracer_provider + ) + + # Verify it's an instance of OpenTelemetry + self.assertIsInstance(levo_logger, LevoLogger) + # Check it extends OpenTelemetry by checking base classes + from litellm.integrations.opentelemetry import OpenTelemetry + + self.assertIsInstance(levo_logger, OpenTelemetry) + + # Verify callback_name is set + self.assertEqual(levo_logger.callback_name, "levo") + + @patch.dict( + "os.environ", + { + "LEVOAI_API_KEY": "test-api-key", + "LEVOAI_ORG_ID": "test-org-id", + "LEVOAI_WORKSPACE_ID": "test-workspace-id", + "LEVOAI_COLLECTOR_URL": "https://collector.levo.ai", + }, + ) + @pytest.mark.skipif( + not OPENTELEMETRY_AVAILABLE, reason="OpenTelemetry packages not installed" + ) + @patch( + "litellm.integrations.opentelemetry.OpenTelemetry._init_otel_logger_on_litellm_proxy" + ) + @pytest.mark.asyncio + async def test_levo_logger_health_check_healthy(self, mock_init_proxy): + """Test health check returns healthy status when config is valid.""" + # Mock the proxy initialization to avoid importing proxy code + mock_init_proxy.return_value = None + + config = LevoLogger.get_levo_config() + otel_config = OpenTelemetryConfig( + exporter=config.protocol, + endpoint=config.endpoint, + headers=config.otlp_auth_headers, + ) + + # Create tracer provider with in-memory exporter + tracer_provider = TracerProvider() + tracer_provider.add_span_processor(SimpleSpanProcessor(InMemorySpanExporter())) + + levo_logger = LevoLogger( + config=otel_config, callback_name="levo", tracer_provider=tracer_provider + ) + + # Run health check + result = await levo_logger.async_health_check() + + self.assertEqual(result["status"], "healthy") + self.assertIn("message", result) + + @patch.dict("os.environ", {}, clear=True) + def test_levo_logger_health_check_unhealthy(self): + """Test health check returns unhealthy status when required vars are missing.""" + # Try to create logger without required env vars + # This should fail during config, but we can test health check logic + with pytest.raises(ValueError): + LevoLogger.get_levo_config() + + @patch.dict( + "os.environ", + { + "LEVOAI_API_KEY": "test-api-key", + "LEVOAI_ORG_ID": "test-org-id", + "LEVOAI_WORKSPACE_ID": "test-workspace-id", + "LEVOAI_COLLECTOR_URL": "https://collector.levo.ai", + }, + ) + @pytest.mark.skipif( + not OPENTELEMETRY_AVAILABLE, reason="OpenTelemetry packages not installed" + ) + @patch( + "litellm.integrations.opentelemetry.OpenTelemetry._init_otel_logger_on_litellm_proxy" + ) + def test_levo_logger_callback_name(self, mock_init_proxy): + """Test that callback_name is properly set and used.""" + # Mock the proxy initialization to avoid importing proxy code + mock_init_proxy.return_value = None + + config = LevoLogger.get_levo_config() + otel_config = OpenTelemetryConfig( + exporter=config.protocol, + endpoint=config.endpoint, + headers=config.otlp_auth_headers, + ) + + # Create tracer provider with in-memory exporter + tracer_provider = TracerProvider() + tracer_provider.add_span_processor(SimpleSpanProcessor(InMemorySpanExporter())) + + levo_logger = LevoLogger( + config=otel_config, callback_name="levo", tracer_provider=tracer_provider + ) + + # Verify callback_name attribute + self.assertEqual(levo_logger.callback_name, "levo") + + +@pytest.mark.parametrize( + "env_vars, expected_headers_contains, expected_endpoint, expected_protocol", + [ + pytest.param( + { + "LEVOAI_API_KEY": "test-key", + "LEVOAI_ORG_ID": "test-org", + "LEVOAI_WORKSPACE_ID": "test-workspace", + "LEVOAI_COLLECTOR_URL": "https://collector.levo.ai", + }, + [ + "Authorization=Bearer test-key", + "x-levo-organization-id=test-org", + "x-levo-workspace-id=test-workspace", + ], + "https://collector.levo.ai", + "otlp_http", + id="collector URL with all required vars", + ), + pytest.param( + { + "LEVOAI_API_KEY": "key-123", + "LEVOAI_ORG_ID": "org-456", + "LEVOAI_WORKSPACE_ID": "workspace-789", + "LEVOAI_COLLECTOR_URL": "https://custom.example.com", + }, + [ + "Authorization=Bearer key-123", + "x-levo-organization-id=org-456", + "x-levo-workspace-id=workspace-789", + ], + "https://custom.example.com", + "otlp_http", + id="custom collector URL", + ), + pytest.param( + { + "LEVOAI_API_KEY": "key-123", + "LEVOAI_ORG_ID": "org-456", + "LEVOAI_WORKSPACE_ID": "workspace-789", + "LEVOAI_COLLECTOR_URL": "http://localhost:9999", + }, + ["Authorization=Bearer key-123"], + "http://localhost:9999", + "otlp_http", + id="custom HTTP endpoint", + ), + ], +) +def test_get_levo_config_parametrized( + monkeypatch, + env_vars, + expected_headers_contains, + expected_endpoint, + expected_protocol, +): + """Parametrized tests for get_levo_config() with various configurations.""" + # Clear all Levo-related env vars first to ensure clean state + for key in [ + "LEVOAI_API_KEY", + "LEVOAI_ORG_ID", + "LEVOAI_WORKSPACE_ID", + "LEVOAI_COLLECTOR_URL", + "LEVOAI_ENV_NAME", + ]: + monkeypatch.delenv(key, raising=False) + + for key, value in env_vars.items(): + monkeypatch.setenv(key, value) + + config = LevoLogger.get_levo_config() + + assert isinstance(config, LevoConfig) + assert config.endpoint == expected_endpoint + assert config.protocol == expected_protocol + + # Verify all expected header parts are present + for header_part in expected_headers_contains: + assert header_part in config.otlp_auth_headers + + +@pytest.mark.parametrize( + "missing_var", + [ + pytest.param("LEVOAI_API_KEY", id="missing API key"), + pytest.param("LEVOAI_ORG_ID", id="missing org ID"), + pytest.param("LEVOAI_WORKSPACE_ID", id="missing workspace ID"), + pytest.param("LEVOAI_COLLECTOR_URL", id="missing collector URL"), + ], +) +def test_get_levo_config_missing_required_vars(monkeypatch, missing_var): + """Test that missing required environment variables raise ValueError.""" + # Clear all Levo-related env vars + for key in [ + "LEVOAI_API_KEY", + "LEVOAI_ORG_ID", + "LEVOAI_WORKSPACE_ID", + "LEVOAI_COLLECTOR_URL", + ]: + monkeypatch.delenv(key, raising=False) + + # Set all required vars except the missing one + required_vars = { + "LEVOAI_API_KEY": "test-key", + "LEVOAI_ORG_ID": "test-org", + "LEVOAI_WORKSPACE_ID": "test-workspace", + "LEVOAI_COLLECTOR_URL": "https://collector.levo.ai", + } + required_vars.pop(missing_var) + + for key, value in required_vars.items(): + monkeypatch.setenv(key, value) + + with pytest.raises(ValueError, match=missing_var): + LevoLogger.get_levo_config() + + +if __name__ == "__main__": + unittest.main()