mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-08 03:08:45 +00:00
[Feat] s3 logger, add support for ssl_verify when using minio logger (#16211)
* fixes s3_v2 verify * test_s3_verify_false_async_client * fix * ruff fixes
This commit is contained in:
parent
df6e084984
commit
0737cc7c13
7 changed files with 222 additions and 25 deletions
|
|
@ -1,6 +1,5 @@
|
|||
import json
|
||||
from abc import ABC
|
||||
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Type, Union
|
||||
from typing import TYPE_CHECKING, Any, Dict, Optional, Type
|
||||
|
||||
from typing_extensions import override
|
||||
|
||||
|
|
@ -101,7 +100,6 @@ def set_attributes(
|
|||
SpanAttributes,
|
||||
ToolCallAttributes,
|
||||
)
|
||||
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
|
||||
|
||||
try:
|
||||
optional_params = kwargs.get("optional_params", {})
|
||||
|
|
|
|||
|
|
@ -5,13 +5,11 @@ Relevant Issue: https://github.com/BerriAI/litellm/issues/13764
|
|||
"""
|
||||
|
||||
import json
|
||||
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
|
||||
from typing import TYPE_CHECKING, Any, Dict, Optional, Union
|
||||
|
||||
from numpy import isin
|
||||
from pydantic import BaseModel
|
||||
from typing_extensions import override
|
||||
|
||||
import litellm
|
||||
from litellm.integrations.opentelemetry_utils.base_otel_llm_obs_attributes import (
|
||||
BaseLLMObsOTELAttributes,
|
||||
safe_set_attribute,
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
from abc import ABC
|
||||
from typing import TYPE_CHECKING, Any, Dict, List, Union
|
||||
from typing import TYPE_CHECKING, Any, Dict, Union
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from opentelemetry.trace import Span
|
||||
|
|
|
|||
|
|
@ -57,12 +57,7 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
|
|||
f"in init s3 logger - s3_callback_params {litellm.s3_callback_params}"
|
||||
)
|
||||
|
||||
# IMPORTANT: We use a concurrent limit of 1 to upload to s3
|
||||
# Files should get uploaded BUT they should not impact latency of LLM calling logic
|
||||
self.async_httpx_client = get_async_httpx_client(
|
||||
llm_provider=httpxSpecialProvider.LoggingCallback,
|
||||
)
|
||||
|
||||
# Initialize S3 params first to get the correct s3_verify value
|
||||
self._init_s3_params(
|
||||
s3_bucket_name=s3_bucket_name,
|
||||
s3_region_name=s3_region_name,
|
||||
|
|
@ -85,6 +80,16 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
|
|||
)
|
||||
verbose_logger.debug(f"s3 logger using endpoint url {s3_endpoint_url}")
|
||||
|
||||
# IMPORTANT
|
||||
# Create httpx client AFTER _init_s3_params so we have the correct s3_verify value
|
||||
verbose_logger.debug(
|
||||
f"s3_v2 logger creating async httpx client with s3_verify={self.s3_verify}"
|
||||
)
|
||||
self.async_httpx_client = get_async_httpx_client(
|
||||
llm_provider=httpxSpecialProvider.LoggingCallback,
|
||||
params={"ssl_verify": self.s3_verify}
|
||||
)
|
||||
|
||||
asyncio.create_task(self.periodic_flush())
|
||||
self.flush_lock = asyncio.Lock()
|
||||
|
||||
|
|
@ -147,9 +152,11 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
|
|||
litellm.s3_callback_params.get("s3_api_version") or s3_api_version
|
||||
)
|
||||
self.s3_use_ssl = (
|
||||
litellm.s3_callback_params.get("s3_use_ssl", True) or s3_use_ssl
|
||||
litellm.s3_callback_params.get("s3_use_ssl", True) if litellm.s3_callback_params.get("s3_use_ssl") is not None else s3_use_ssl
|
||||
)
|
||||
self.s3_verify = (
|
||||
litellm.s3_callback_params.get("s3_verify") if litellm.s3_callback_params.get("s3_verify") is not None else s3_verify
|
||||
)
|
||||
self.s3_verify = litellm.s3_callback_params.get("s3_verify") or s3_verify
|
||||
self.s3_endpoint_url = (
|
||||
litellm.s3_callback_params.get("s3_endpoint_url") or s3_endpoint_url
|
||||
)
|
||||
|
|
@ -279,6 +286,9 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
|
|||
verbose_logger.debug(
|
||||
f"s3_v2 logger - uploading data to s3 - {batch_logging_element.s3_object_key}"
|
||||
)
|
||||
verbose_logger.debug(
|
||||
f"s3_v2 logger - s3_verify setting: {self.s3_verify}"
|
||||
)
|
||||
|
||||
# Prepare the URL
|
||||
url = f"https://{self.s3_bucket_name}.s3.{self.s3_region_name}.amazonaws.com/{batch_logging_element.s3_object_key}"
|
||||
|
|
@ -477,7 +487,9 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
|
|||
# Prepare the signed headers
|
||||
signed_headers = dict(aws_request.headers.items())
|
||||
|
||||
httpx_client = _get_httpx_client()
|
||||
httpx_client = _get_httpx_client(
|
||||
params={"ssl_verify": self.s3_verify} if self.s3_verify is not None else None
|
||||
)
|
||||
# Make the request
|
||||
response = httpx_client.put(url, data=json_string, headers=signed_headers)
|
||||
response.raise_for_status()
|
||||
|
|
|
|||
|
|
@ -9,7 +9,6 @@ from typing import Any, Dict, Optional
|
|||
|
||||
from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
|
||||
from litellm.llms.vertex_ai.vertex_llm_base import VertexBase
|
||||
from litellm.types.llms.vertex_ai import VertexPartnerProvider
|
||||
|
||||
|
||||
class VertexAIPartnerModelsTokenCounter(VertexBase):
|
||||
|
|
|
|||
|
|
@ -23,12 +23,20 @@ search_tools:
|
|||
api_key: os.environ/EXA_API_KEY
|
||||
|
||||
|
||||
|
||||
# for /files endpoints
|
||||
files_settings:
|
||||
- custom_llm_provider: openai
|
||||
api_key: os.environ/OPENAI_API_KEY
|
||||
|
||||
|
||||
litellm_settings:
|
||||
callbacks: ["datadog"]
|
||||
# Comprehensive logging settings
|
||||
store_audit_logs: true
|
||||
verbose: true
|
||||
log_level: "DEBUG" # Options: DEBUG, INFO, WARNING, ERROR
|
||||
success_callback: ["s3_v2"]
|
||||
s3_callback_params:
|
||||
s3_endpoint_url: "https://localhost:443" # Replace with your Minio server URL and port
|
||||
s3_aws_access_key_id: "minioadmin"
|
||||
s3_aws_secret_access_key: "minioadmin"
|
||||
s3_region_name: "minio" # This can be any value for Minio
|
||||
s3_bucket_name: "litellm-test" # Replace with your bucket name
|
||||
s3_use_ssl: False
|
||||
s3_verify: False
|
||||
cache: True
|
||||
cache_params:
|
||||
type: local
|
||||
|
|
@ -30,6 +30,7 @@ class TestS3V2UnitTests:
|
|||
def test_s3_v2_endpoint_url(self, mock_periodic_flush, mock_create_task):
|
||||
"""testing s3 endpoint url"""
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
from litellm.types.integrations.s3_v2 import s3BatchLoggingElement
|
||||
|
||||
# Mock periodic_flush and create_task to prevent async task creation during init
|
||||
|
|
@ -266,6 +267,187 @@ async def test_strip_base64_mixed_nested_objects():
|
|||
assert stripped["messages"][0]["extra"]["trace_id"] == "123"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_s3_verify_false_handling():
|
||||
"""
|
||||
Test that s3_verify=False is properly handled and not treated as None.
|
||||
|
||||
This is a regression test for the bug where s3_verify=False was being
|
||||
ignored because 'False or s3_verify' would evaluate to s3_verify (None).
|
||||
"""
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import litellm
|
||||
|
||||
# Set up s3_callback_params with s3_verify=False
|
||||
litellm.s3_callback_params = {
|
||||
"s3_bucket_name": "test-bucket",
|
||||
"s3_endpoint_url": "https://localhost:443",
|
||||
"s3_aws_access_key_id": "minioadmin",
|
||||
"s3_aws_secret_access_key": "minioadmin",
|
||||
"s3_region_name": "us-east-1",
|
||||
"s3_verify": False, # This should NOT be ignored
|
||||
"s3_use_ssl": False, # This should also NOT be ignored
|
||||
}
|
||||
|
||||
with patch('asyncio.create_task'):
|
||||
with patch('litellm.integrations.s3_v2.get_async_httpx_client') as mock_get_client:
|
||||
mock_client = AsyncMock()
|
||||
mock_get_client.return_value = mock_client
|
||||
|
||||
# Create logger
|
||||
logger = S3Logger()
|
||||
|
||||
# Verify s3_verify is False, not None
|
||||
assert logger.s3_verify is False, f"Expected s3_verify=False, got {logger.s3_verify}"
|
||||
assert logger.s3_use_ssl is False, f"Expected s3_use_ssl=False, got {logger.s3_use_ssl}"
|
||||
|
||||
# Verify that get_async_httpx_client was called with ssl_verify=False
|
||||
mock_get_client.assert_called_once()
|
||||
call_kwargs = mock_get_client.call_args.kwargs
|
||||
assert 'params' in call_kwargs, "params should be passed to get_async_httpx_client"
|
||||
assert call_kwargs['params'] == {'ssl_verify': False}, f"Expected ssl_verify=False in params, got {call_kwargs.get('params')}"
|
||||
|
||||
# Clean up
|
||||
litellm.s3_callback_params = None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_s3_verify_none_handling():
|
||||
"""
|
||||
Test that s3_verify=None uses default behavior.
|
||||
"""
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import litellm
|
||||
|
||||
# Set up s3_callback_params without s3_verify
|
||||
litellm.s3_callback_params = {
|
||||
"s3_bucket_name": "test-bucket",
|
||||
"s3_aws_access_key_id": "test-key",
|
||||
"s3_aws_secret_access_key": "test-secret",
|
||||
"s3_region_name": "us-east-1",
|
||||
}
|
||||
|
||||
with patch('asyncio.create_task'):
|
||||
with patch('litellm.integrations.s3_v2.get_async_httpx_client') as mock_get_client:
|
||||
mock_client = AsyncMock()
|
||||
mock_get_client.return_value = mock_client
|
||||
|
||||
# Create logger without explicit s3_verify
|
||||
logger = S3Logger()
|
||||
|
||||
# Verify s3_verify is None (default)
|
||||
assert logger.s3_verify is None, f"Expected s3_verify=None, got {logger.s3_verify}"
|
||||
|
||||
# Verify that get_async_httpx_client was called
|
||||
mock_get_client.assert_called_once()
|
||||
call_kwargs = mock_get_client.call_args.kwargs
|
||||
# When s3_verify is None, params={'ssl_verify': None} which is fine - uses default behavior
|
||||
# The important thing is it's not False
|
||||
if 'params' in call_kwargs and call_kwargs['params'] is not None:
|
||||
assert call_kwargs['params'].get('ssl_verify') is None
|
||||
# Either params is None or params={'ssl_verify': None} is acceptable
|
||||
|
||||
# Clean up
|
||||
litellm.s3_callback_params = None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_s3_verify_false_creates_httpx_client_with_verify_false():
|
||||
"""
|
||||
Test that when s3_verify=False, the actual httpx client has verify=False.
|
||||
|
||||
This validates that ssl_verify=False flows through to the httpx.AsyncClient.
|
||||
"""
|
||||
from unittest.mock import patch
|
||||
|
||||
import litellm
|
||||
|
||||
# Set up s3_callback_params with s3_verify=False
|
||||
litellm.s3_callback_params = {
|
||||
"s3_bucket_name": "test-bucket",
|
||||
"s3_endpoint_url": "https://localhost:443",
|
||||
"s3_aws_access_key_id": "minioadmin",
|
||||
"s3_aws_secret_access_key": "minioadmin",
|
||||
"s3_region_name": "us-east-1",
|
||||
"s3_verify": False,
|
||||
}
|
||||
|
||||
with patch('asyncio.create_task'):
|
||||
# Create logger - this creates the httpx client
|
||||
logger = S3Logger()
|
||||
|
||||
# Verify the logger has s3_verify=False
|
||||
assert logger.s3_verify is False
|
||||
|
||||
# Check the actual httpx client has verify=False
|
||||
# The async_httpx_client.client is the actual httpx.AsyncClient
|
||||
if hasattr(logger.async_httpx_client, 'client'):
|
||||
httpx_client = logger.async_httpx_client.client
|
||||
# Check the _verify attribute (httpx internal)
|
||||
if hasattr(httpx_client, '_verify'):
|
||||
assert httpx_client._verify is False, f"Expected httpx client _verify=False, got {httpx_client._verify}"
|
||||
|
||||
# Clean up
|
||||
litellm.s3_callback_params = None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_s3_verify_false_async_client():
|
||||
"""
|
||||
Test that the async httpx client respects s3_verify=False.
|
||||
"""
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import litellm
|
||||
from litellm.types.integrations.s3_v2 import s3BatchLoggingElement
|
||||
|
||||
# Set up s3_callback_params with s3_verify=False
|
||||
litellm.s3_callback_params = {
|
||||
"s3_bucket_name": "test-bucket",
|
||||
"s3_endpoint_url": "https://localhost:443",
|
||||
"s3_aws_access_key_id": "minioadmin",
|
||||
"s3_aws_secret_access_key": "minioadmin",
|
||||
"s3_region_name": "us-east-1",
|
||||
"s3_verify": False,
|
||||
}
|
||||
|
||||
with patch('asyncio.create_task'):
|
||||
logger = S3Logger()
|
||||
|
||||
# Verify s3_verify is False
|
||||
assert logger.s3_verify is False
|
||||
|
||||
# Create test element
|
||||
test_element = s3BatchLoggingElement(
|
||||
s3_object_key="2025-11-03/test-key.json",
|
||||
payload={"test": "data"},
|
||||
s3_object_download_filename="test-file.json"
|
||||
)
|
||||
|
||||
# Mock the async httpx client's put method
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
logger.async_httpx_client.put = AsyncMock(return_value=mock_response)
|
||||
|
||||
# Call async upload
|
||||
await logger.async_upload_data_to_s3(test_element)
|
||||
|
||||
# Verify put was called
|
||||
assert logger.async_httpx_client.put.called
|
||||
|
||||
# Check that the async httpx client was created with verify=False
|
||||
if hasattr(logger.async_httpx_client, 'client'):
|
||||
httpx_client = logger.async_httpx_client.client
|
||||
if hasattr(httpx_client, '_verify'):
|
||||
assert httpx_client._verify is False, f"Expected async httpx client _verify=False, got {httpx_client._verify}"
|
||||
|
||||
# Clean up
|
||||
litellm.s3_callback_params = None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_strip_base64_recursive_redaction():
|
||||
logger = S3Logger(s3_strip_base64_files=True)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue