From bfd3f39dca061459819939d534d54b48b91eed1f Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 1 Oct 2026 11:00:21 -0700 Subject: [PATCH] feat(s3_v2): add s3_partition_granularity option for hourly S3 folders (#43748) * feat(s3_v2): add s3_partition_granularity option for hourly S3 folders Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): cover s3 v2 partition granularity across surfaces, settings and chaos Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): cover previous_response_id history rebuilt from an hourly cold storage object Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(s3_v2): reuse the cold storage key only when s3_v2 owns cold storage Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(s3_v2): cover hour rollover, postgres outage, in-flight switches, key/team vars and real S3 layout Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * chore(liccheck): authorize libfaketime, the GPLv2 dev-only clock the s3 rollover integration test preloads Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(s3_v2): wait for the rejected-request cell's payloads by id, not by line count Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(s3_v2): declare the postgres outage cell's models in config and trip the relay on burst ids Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(s3_v2): drop the libfaketime hour rollover cell and its dev dependency Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(e2e): deselect the s3_v2 live e2e on the stage-mirror stack The stage-mirror config enables no s3_v2 callback, so every test in test_s3_log_e2e.py fails its readiness check there. The file keeps running in the Buildkite e2e lane, which configures s3_v2 Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(s3_v2): declare the sink outage burst models in config Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * refactor(s3_v2): read cold storage metadata without an empty dict default Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): wait for the proxy to reconnect before the postgres outage recovery request Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: mrinal Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Co-authored-by: yucheng --- .github/e2e-stack/select_tests.py | 1 + litellm/constants.py | 1 + litellm/integrations/callback_configs.json | 7 + litellm/integrations/s3.py | 17 +- litellm/integrations/s3_v2.py | 55 +- litellm/litellm_core_utils/litellm_logging.py | 5 + litellm/proxy/_types.py | 1 + litellm/types/integrations/s3_v2.py | 4 + tests/e2e/coverage_registry/logging.yaml | 1 + tests/e2e/e2e_config.py | 1 + tests/e2e/logging/test_s3_log_e2e.py | 43 +- tests/integration/_support/database_relay.py | 3 + .../test_s3_v2_partition_granularity.py | 1241 +++++++++++++++++ tests/unit/integrations/test_s3.py | 7 + tests/unit/integrations/test_s3_v2.py | 247 ++++ .../test_litellm_logging.py | 3 + .../src/components/settings.test.tsx | 47 + .../src/components/settings.tsx | 4 +- 18 files changed, 1679 insertions(+), 9 deletions(-) create mode 100644 tests/integration/observability/test_s3_v2_partition_granularity.py diff --git a/.github/e2e-stack/select_tests.py b/.github/e2e-stack/select_tests.py index e425c313d6a..792da5ae09c 100644 --- a/.github/e2e-stack/select_tests.py +++ b/.github/e2e-stack/select_tests.py @@ -11,6 +11,7 @@ UNSUPPORTED: Final = re.compile( r"|^tests/e2e/guardrails/test_presidio_masking_e2e\.py$" r"|^tests/e2e/logging/test_otel_v2_langfuse_generation_output_e2e\.py$" r"|^tests/e2e/logging/test_langsmith_batch_serialization_e2e\.py$" + r"|^tests/e2e/logging/test_s3_log_e2e\.py$" r"|^tests/e2e/secret_manager/" ) HARNESS: Final = re.compile( diff --git a/litellm/constants.py b/litellm/constants.py index b3f5b0471f4..7e1e63a112b 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -69,6 +69,7 @@ S3_PREFIX_DIGEST_CHARS: Final = 16 # s3 allows 2048 bytes of combined metadata headers, which Content-Disposition counts against MAX_S3_OBJECT_DOWNLOAD_FILENAME_BYTES: Final = 1024 S3_LOG_PROMPTS_ONLY_ENV_VAR: Final = "S3_LOG_PROMPTS_ONLY" +S3_PARTITION_GRANULARITY_ENV_VAR: Final = "S3_PARTITION_GRANULARITY" MAX_FILE_LIST_LIMIT: Final = 10000 DEFAULT_SQS_FLUSH_INTERVAL_SECONDS: Final = int(os.getenv("DEFAULT_SQS_FLUSH_INTERVAL_SECONDS", 10)) DEFAULT_NUM_WORKERS_LITELLM_PROXY: Final = int(os.getenv("DEFAULT_NUM_WORKERS_LITELLM_PROXY", 1)) diff --git a/litellm/integrations/callback_configs.json b/litellm/integrations/callback_configs.json index 190c283d087..38928f67f42 100644 --- a/litellm/integrations/callback_configs.json +++ b/litellm/integrations/callback_configs.json @@ -498,6 +498,13 @@ "ui_name": "Log Prompts Only", "description": "Log request messages to S3 but drop the model response from each logged object", "required": false + }, + "s3_partition_granularity": { + "type": "select", + "ui_name": "Folder Partitioning", + "description": "day writes one folder per date, hour adds an hour folder below each date (s3_v2 only)", + "options": ["day", "hour"], + "required": false } }, "description": "S3 Bucket (AWS) Logging Integration" diff --git a/litellm/integrations/s3.py b/litellm/integrations/s3.py index f330ca8e0ac..129fceb40bf 100644 --- a/litellm/integrations/s3.py +++ b/litellm/integrations/s3.py @@ -16,8 +16,10 @@ from litellm.constants import ( MAX_S3_OBJECT_KEY_BYTES, S3_BOUNDED_OBJECT_KEY_HEAD_BYTES, S3_LOG_PROMPTS_ONLY_ENV_VAR, + S3_PARTITION_GRANULARITY_ENV_VAR, S3_PREFIX_DIGEST_CHARS, ) +from litellm.types.integrations.s3_v2 import S3PartitionGranularity from litellm.types.utils import StandardLoggingPayload _S3_BOOL: Final = TypeAdapter(bool) @@ -36,6 +38,18 @@ def resolve_s3_log_prompts_only(configured: object, environ: Mapping[str, str] | return True +def resolve_s3_partition_granularity( + configured: object, environ: Mapping[str, str] | None = None +) -> S3PartitionGranularity: + env: Final = os.environ if environ is None else environ + raw: Final = env.get(S3_PARTITION_GRANULARITY_ENV_VAR) if configured is None else configured + if raw == "hour": + return "hour" + if raw is not None and raw not in ("", "day"): + verbose_logger.warning("s3 logging: s3_partition_granularity=%r is not one of day, hour, using day", raw) + return "day" + + def _resolve_positive_int(setting: str, configured: object, fallback: int, *, reject_bool: bool) -> int: if configured is None or configured == "": return fallback @@ -371,10 +385,11 @@ def get_s3_object_key( prefix: str, start_time: datetime, s3_file_name: str, + partition_granularity: S3PartitionGranularity = "day", ) -> str: sanitized_s3_file_name: Final = s3_file_name.replace("/", "_").replace(":", "_") configured_prefix: Final = (s3_path.rstrip("/") + "/" if s3_path else "") + prefix - date_segment: Final = start_time.strftime("%Y-%m-%d") + "/" + date_segment: Final = start_time.strftime("%Y-%m-%d/%H/" if partition_granularity == "hour" else "%Y-%m-%d/") # we need the s3 key to include the time, so we log cache hits too s3_object_key: Final = configured_prefix + date_segment + sanitized_s3_file_name + ".json" if len(s3_object_key.encode("utf-8")) <= MAX_S3_OBJECT_KEY_BYTES: diff --git a/litellm/integrations/s3_v2.py b/litellm/integrations/s3_v2.py index 88d7906cc4b..ea9e6a84c93 100644 --- a/litellm/integrations/s3_v2.py +++ b/litellm/integrations/s3_v2.py @@ -9,6 +9,7 @@ NOTE 1: S3 does not provide a BATCH PUT API endpoint; by default each element is import asyncio import contextvars import logging +import os import re import time from collections.abc import Awaitable, Callable, Mapping @@ -28,6 +29,7 @@ from litellm.constants import ( DEFAULT_S3_FLUSH_INTERVAL_SECONDS, DEFAULT_S3_MAX_ADAPTIVE_CONCURRENCY, DEFAULT_S3_MAX_CONCURRENT_UPLOADS, + S3_PARTITION_GRANULARITY_ENV_VAR, ) from litellm.integrations.adaptive_concurrency import AdaptiveConcurrencyLimiter, PutSample from litellm.integrations.s3 import ( @@ -42,6 +44,7 @@ from litellm.integrations.s3 import ( resolve_s3_max_concurrent_uploads, resolve_s3_max_queue_size, resolve_s3_max_retry_age_seconds, + resolve_s3_partition_granularity, resolve_sse_params, ) from litellm.litellm_core_utils.aws_partition import get_aws_dns_suffix @@ -53,7 +56,7 @@ from litellm.llms.custom_httpx.http_handler import ( get_async_httpx_client, httpxSpecialProvider, ) -from litellm.types.integrations.s3_v2 import s3BatchLoggingElement +from litellm.types.integrations.s3_v2 import S3PartitionGranularity, s3BatchLoggingElement from litellm.types.utils import StandardAuditLogPayload, StandardLoggingPayload from .custom_batch_logger import CustomBatchLogger @@ -119,6 +122,8 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM): _upload_limiter: asyncio.Semaphore | AdaptiveConcurrencyLimiter | None = None s3_drop_on_terminal_error: bool = True s3_max_retry_age_seconds: int | None = 3600 + s3_partition_granularity: object = None + _partition_granularity_cache: tuple[object, S3PartitionGranularity] | None = None def __init__( self, @@ -147,6 +152,7 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM): s3_server_side_encryption: str | None = None, s3_sse_kms_key_id: str | None = None, s3_log_prompts_only: bool | None = None, + s3_partition_granularity: str | None = None, s3_max_concurrent_uploads: int = DEFAULT_S3_MAX_CONCURRENT_UPLOADS, s3_max_queue_size: int | None = None, s3_max_retry_age_seconds: int | None = 3600, @@ -195,6 +201,7 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM): s3_server_side_encryption=s3_server_side_encryption, s3_sse_kms_key_id=s3_sse_kms_key_id, s3_log_prompts_only=s3_log_prompts_only, + s3_partition_granularity=s3_partition_granularity, s3_max_concurrent_uploads=s3_max_concurrent_uploads, s3_max_queue_size=s3_max_queue_size, s3_max_retry_age_seconds=s3_max_retry_age_seconds, @@ -271,6 +278,7 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM): s3_server_side_encryption: str | None = None, s3_sse_kms_key_id: str | None = None, s3_log_prompts_only: bool | None = None, + s3_partition_granularity: str | None = None, s3_max_concurrent_uploads: int = DEFAULT_S3_MAX_CONCURRENT_UPLOADS, s3_max_queue_size: int | None = None, s3_max_retry_age_seconds: int | None = 3600, @@ -331,6 +339,11 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM): params.get("s3_log_prompts_only") if s3_log_prompts_only is None else s3_log_prompts_only ) + self.s3_partition_granularity = ( + params.get("s3_partition_granularity") if s3_partition_granularity is None else s3_partition_granularity + ) + self._partition_granularity_cache = None + self.s3_server_side_encryption, self.s3_sse_kms_key_id = resolve_sse_params( params.get("s3_server_side_encryption") or s3_server_side_encryption, params.get("s3_sse_kms_key_id") or s3_sse_kms_key_id, @@ -482,6 +495,7 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM): "audit_logs/", now, f"{now.strftime('%H-%M-%S')}_{audit_log_id}", + partition_granularity=self.resolve_partition_granularity(), ) element: Final = s3BatchLoggingElement( @@ -758,6 +772,19 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM): ), ) + def resolve_partition_granularity(self) -> S3PartitionGranularity: + raw: Final = ( + os.environ.get(S3_PARTITION_GRANULARITY_ENV_VAR) + if self.s3_partition_granularity is None + else self.s3_partition_granularity + ) + cached: Final = self._partition_granularity_cache + if cached is not None and cached[0] == raw: + return cached[1] + resolved: Final = resolve_s3_partition_granularity(raw) + self._partition_granularity_cache = (raw, resolved) + return resolved + def create_s3_batch_logging_element( self, start_time: datetime, @@ -803,11 +830,27 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM): prefix_path, s3_file_name, ) - s3_object_key: Final = get_s3_object_key( - s3_path=cast(str | None, self.s3_path) or "", - prefix=prefix_path, - start_time=start_time, - s3_file_name=s3_file_name, + + def object_key(partition_granularity: S3PartitionGranularity) -> str: + return get_s3_object_key( + s3_path=cast(str | None, self.s3_path) or "", + prefix=prefix_path, + start_time=start_time, + s3_file_name=s3_file_name, + partition_granularity=partition_granularity, + ) + + metadata: Final = standard_logging_payload.get("metadata") + cold_storage_object_key: Final = ( + metadata.get("cold_storage_object_key") + if metadata is not None and litellm.cold_storage_custom_logger == "s3_v2" + else None + ) + s3_object_key: Final = ( + cold_storage_object_key + if cold_storage_object_key is not None + and cold_storage_object_key in (object_key("day"), object_key("hour")) + else object_key(self.resolve_partition_granularity()) ) verbose_logger.debug("s3_object_key=%s", s3_object_key) diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 154893b6c21..097ca2bcd54 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -117,6 +117,7 @@ from litellm.llms.base_llm.search.transformation import SearchResponse from litellm.responses.utils import ResponseAPILoggingUtils from litellm.types.agents import LiteLLMSendMessageResponse from litellm.types.containers.main import ContainerObject +from litellm.types.integrations.s3_v2 import S3PartitionGranularity from litellm.types.interactions import ( InteractionsAPIResponse, InteractionsAPIStreamingResponse, @@ -6059,6 +6060,7 @@ class StandardLoggingPayloadSetup: # Get the actual s3_path from the configured cold storage logger instance s3_path = "" # default value + partition_granularity: S3PartitionGranularity = "day" # Try to get the actual logger instance from the logger name try: @@ -6067,6 +6069,8 @@ class StandardLoggingPayloadSetup: ) if custom_logger and hasattr(custom_logger, "s3_path") and getattr(custom_logger, "s3_path"): s3_path = getattr(custom_logger, "s3_path") + if isinstance(custom_logger, S3V2Logger): + partition_granularity = custom_logger.resolve_partition_granularity() except Exception: # If any error occurs in getting the logger instance, use default empty s3_path pass @@ -6076,6 +6080,7 @@ class StandardLoggingPayloadSetup: prefix="", # Don't split by team alias for cold storage start_time=start_time, s3_file_name=s3_file_name, + partition_granularity=partition_granularity, ) return s3_object_key diff --git a/litellm/proxy/_types.py b/litellm/proxy/_types.py index 649381f24a2..bb3c11f01e4 100644 --- a/litellm/proxy/_types.py +++ b/litellm/proxy/_types.py @@ -3957,6 +3957,7 @@ class AllCallbacks(LiteLLMPydanticObjectBase): "AWS_SECRET_ACCESS_KEY", "AWS_REGION_NAME", "S3_LOG_PROMPTS_ONLY", + "S3_PARTITION_GRANULARITY", ], ) diff --git a/litellm/types/integrations/s3_v2.py b/litellm/types/integrations/s3_v2.py index 3b0dad97e8c..e8ad28f1a3b 100644 --- a/litellm/types/integrations/s3_v2.py +++ b/litellm/types/integrations/s3_v2.py @@ -1,5 +1,9 @@ +from typing import Literal + from pydantic import BaseModel +S3PartitionGranularity = Literal["day", "hour"] + class s3BatchLoggingElement(BaseModel): """ diff --git a/tests/e2e/coverage_registry/logging.yaml b/tests/e2e/coverage_registry/logging.yaml index 7c83e4d3aea..5424fb2d12f 100644 --- a/tests/e2e/coverage_registry/logging.yaml +++ b/tests/e2e/coverage_registry/logging.yaml @@ -1,6 +1,7 @@ # Logging integration delivery (behavior features). Grounded in litellm/integrations/. - {id: logging.s3.success.writes_object, module: logging, tier: P0, event: success, assertions: [writes_object], exercised_on: [chat_completions, messages, embeddings], source: "integrations/s3_v2.py", rationale: "Primary audit trail; batch flush no-drop"} - {id: logging.s3.failure.writes_object, module: logging, tier: P0, event: failure, assertions: [writes_object], exercised_on: [chat_completions, messages], source: "integrations/s3_v2.py", rationale: "Failed calls persisted for compliance"} +- {id: logging.s3.success.partition_layout, module: logging, tier: P1, event: success, assertions: [object_key_layout], exercised_on: [chat_completions], source: "integrations/s3_v2.py / LIT-8985", rationale: "s3_partition_granularity picks the date or date/hour folder every downstream query and lifecycle rule reads"} - {id: logging.gcs_bucket.success.writes_object, module: logging, tier: P0, event: success, assertions: [writes_object], exercised_on: [chat_completions, messages, embeddings], source: "integrations/gcs_bucket/gcs_bucket.py", rationale: "GCS parallel to S3"} - {id: logging.datadog.success.exports_metric, module: logging, tier: P0, event: success, assertions: [exports_metric], exercised_on: [chat_completions, messages, responses, embeddings], source: "integrations/datadog/datadog.py", rationale: "Powers dashboards/alerts; cardinality regressions common"} - {id: logging.datadog.stream.exports_metric, module: logging, tier: P0, event: stream, assertions: [exports_metric], exercised_on: [chat_completions, messages, responses], source: "integrations/datadog/datadog.py", rationale: "Streaming aggregates usage after the last chunk; delivery and cost must survive that path"} diff --git a/tests/e2e/e2e_config.py b/tests/e2e/e2e_config.py index b2682c04841..3fa9f534ffd 100644 --- a/tests/e2e/e2e_config.py +++ b/tests/e2e/e2e_config.py @@ -106,6 +106,7 @@ UI_BASE_URL = os.environ.get("E2E_UI_BASE_URL", PROXY_BASE_URL).rstrip("/") CHEAP_ANTHROPIC_MODEL = os.environ.get("E2E_CHEAP_ANTHROPIC_MODEL", "claude-haiku-4-5") CHEAP_OPENAI_MODEL = os.environ.get("E2E_CHEAP_OPENAI_MODEL", "gpt-5.5") +S3_PARTITION_GRANULARITY = os.environ.get("E2E_S3_PARTITION_GRANULARITY", "day") LINEAR_MCP_URL = os.environ.get("E2E_LINEAR_MCP_URL", "https://mcp.linear.app/mcp") LINEAR_STORAGE_STATE = os.environ.get("E2E_LINEAR_STORAGE_STATE", "") diff --git a/tests/e2e/logging/test_s3_log_e2e.py b/tests/e2e/logging/test_s3_log_e2e.py index 7a1ee1e6536..1612fa315a6 100644 --- a/tests/e2e/logging/test_s3_log_e2e.py +++ b/tests/e2e/logging/test_s3_log_e2e.py @@ -22,11 +22,12 @@ alias per test turns the poll into a cheap prefix listing. from __future__ import annotations import math +import re import time import pytest -from e2e_config import CHEAP_ANTHROPIC_MODEL, unique_marker +from e2e_config import CHEAP_ANTHROPIC_MODEL, S3_PARTITION_GRANULARITY, unique_marker from lifecycle import ResourceManager from logging_client import ( INVALID_UPSTREAM_API_KEY, @@ -106,6 +107,46 @@ class TestS3LogDelivery: record.response_cost, outcome.response_cost, rel_tol=1e-9 ), f"payload response_cost {record.response_cost!r} must equal the header cost {outcome.response_cost}" + @pytest.mark.covers("logging.s3.success.partition_layout", exercised_on=["chat_completions"]) + def test_chat_completions_object_key_follows_the_partition_granularity( + self, client: LoggingClient, s3_logs: S3LogReader, resources: ResourceManager + ) -> None: + """The one object a call writes must sit in the folder layout the proxy's + s3_partition_granularity names: {alias}/{date}/ for day and + {alias}/{date}/{HH}/ for hour, where HH is the hour the object's own + time- file name records. E2E_S3_PARTITION_GRANULARITY tells the test + which one the proxy under test runs.""" + _assert_s3_configured(client) + + alias = f"s3-layout-{unique_marker()}" + key = client.key_with_alias(alias, models=[CHEAP_ANTHROPIC_MODEL]) + resources.defer(lambda: client.delete_key(key)) + + outcome = first_ok( + client, + lambda: client.chat_raw( + key, CHEAP_ANTHROPIC_MODEL, f"reply with one word {unique_marker()}", max_tokens=16 + ), + ) + body_id = completion_response_id(outcome.body) + assert body_id is not None, "the completion body must carry an id (it names the s3 object)" + records = s3_logs.poll_records(prefix=f"{alias}/", predicate=lambda r: r.id == body_id) + assert len(records) == 1, f"expected exactly ONE s3 object for response {body_id}, got {len(records)}" + + file_id = body_id.replace("/", "_").replace(":", "_") + hour_folder = r"(?P\d{2})/" if S3_PARTITION_GRANULARITY == "hour" else "" + layout = re.compile( + rf"{re.escape(alias)}/\d{{4}}-\d{{2}}-\d{{2}}/{hour_folder}" + rf"time-(?P\d{{2}})-\d{{2}}-\d{{2}}-\d{{6}}_{re.escape(file_id)}\.json" + ) + keys = [object_key for object_key in s3_logs.list_keys(f"{alias}/") if file_id in object_key] + assert len(keys) == 1, f"expected one object key for response {body_id}, got {keys}" + match = layout.fullmatch(keys[0]) + assert match is not None, f"{keys[0]!r} is outside the {S3_PARTITION_GRANULARITY} layout {layout.pattern!r}" + assert S3_PARTITION_GRANULARITY != "hour" or match.group("folder_hour") == match.group("file_hour"), ( + f"the hour folder must be the hour the object's file name records: {keys[0]!r}" + ) + @pytest.mark.covers("logging.s3.failure.writes_object", exercised_on=["chat_completions"]) def test_chat_completions_failure_writes_one_object( self, client: LoggingClient, s3_logs: S3LogReader, resources: ResourceManager diff --git a/tests/integration/_support/database_relay.py b/tests/integration/_support/database_relay.py index bb0243226a8..1b3bc0183a5 100644 --- a/tests/integration/_support/database_relay.py +++ b/tests/integration/_support/database_relay.py @@ -29,6 +29,7 @@ class DatabaseRelay: self._armed: Final = threading.Event() self.tripped: Final = threading.Event() self.refused = 0 + self.reconnected: Final = threading.Event() self._tripped_at = 0.0 self._writers: tuple[asyncio.StreamWriter, ...] = () self._ready: Final = threading.Event() @@ -61,6 +62,8 @@ class DatabaseRelay: self.refused += 1 client_writer.close() return + if self.tripped.is_set(): + self.reconnected.set() server_reader, server_writer = await asyncio.open_connection(self._upstream_host, self._upstream_port) self._writers = (*self._writers, client_writer, server_writer) diff --git a/tests/integration/observability/test_s3_v2_partition_granularity.py b/tests/integration/observability/test_s3_v2_partition_granularity.py new file mode 100644 index 00000000000..dc0a7184cf8 --- /dev/null +++ b/tests/integration/observability/test_s3_v2_partition_granularity.py @@ -0,0 +1,1241 @@ +import json +import re +import threading +import uuid +from collections.abc import Iterator, Mapping +from concurrent.futures import ThreadPoolExecutor +from contextlib import contextmanager +from dataclasses import dataclass, field +from datetime import datetime, timedelta +from pathlib import Path +from typing import Final +from urllib.parse import quote, unquote + +import httpx +import openai +import psutil +import pytest +import yaml +from _s3_v2_support import ( + BUCKET, + PREFIX, + SURFACES, + RecordingS3Sink, + call_surface, + collect_payloads, + matched_ids, + mixed_burst, + s3_config, + surface_reply, +) +from integration._support.client import Gateway, JsonValue, Scenario, eventually, object_value +from integration._support.database import read_rows, scratch_database +from integration._support.database_relay import database_relay +from integration._support.process import OwnedProxy, group_members, owned_proxy_process +from integration._support.wire import Reply, Request, wire_server + +FLUSH: Final = {"DEFAULT_S3_FLUSH_INTERVAL_SECONDS": "1"} +HOUR: Final = {"s3_partition_granularity": "hour"} +ANTHROPIC_MODEL: Final = "anthropic/claude-sonnet-4-5-20250929" +WARNING: Final = "s3 logging: s3_partition_granularity=" +SINK_CREDENTIALS: Final = { + "s3_bucket_name": BUCKET, + "s3_region_name": "us-east-1", + "s3_path": PREFIX, + "s3_aws_access_key_id": "AKIAIOSFODNN7EXAMPLE", + "s3_aws_secret_access_key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", +} + + +@dataclass(slots=True) +class CountingUpstream: + """Scripted provider that answers every surface and fails any prompt ending in -fail with a 401.""" + + lock: threading.Lock = field(default_factory=threading.Lock) + prompts: list[str] = field(default_factory=list) # mutable-ok: appended per upstream request under lock + + def respond(self, request: Request) -> Reply: + if request.method != "POST" or not request.body: + return Reply(status=404) + body: Final = json.loads(request.body) + prompt: Final = str(body["input"] if "input" in body else body["messages"][0]["content"]) + with self.lock: + self.prompts.append(prompt) + if prompt.endswith("-fail"): + return Reply(status=401, body=b'{"error": {"message": "synthetic upstream rejection", "code": "401"}}') + return surface_reply(request) + + def received(self) -> tuple[str, ...]: + with self.lock: + return tuple(self.prompts) + + +def _prompt(payload: Mapping[str, JsonValue]) -> str: + messages: Final = payload["messages"] + if isinstance(messages, str): + return messages + assert isinstance(messages, list) and len(messages) == 1, payload + first: Final = messages[0] + return first if isinstance(first, str) else str(object_value(first)["content"]) + + +def _start(payload: Mapping[str, JsonValue]) -> datetime: + return datetime.fromtimestamp(float(str(payload["startTime"]))) + + +def _folder(payload: Mapping[str, JsonValue], granularity: str, prefix: str = "") -> str: + start: Final = _start(payload) + hour: Final = f"{start:%H}/" if granularity == "hour" else "" + return f"/{BUCKET}/{PREFIX}/{prefix}{start:%Y-%m-%d}/{hour}" + + +def _object_pattern(payload: Mapping[str, JsonValue], granularity: str, prefix: str = "") -> re.Pattern[str]: + return re.compile( + re.escape(_folder(payload, granularity, prefix)) + rf"time-{_start(payload):%H-%M-%S}-\d{{6}}_[^/]+\.json" + ) + + +def _outside_layout(objects: Mapping[str, bytes], granularity: str, prefix: str = "") -> tuple[str, ...]: + return tuple( + target + for target, body in objects.items() + if not _object_pattern(object_value(json.loads(body)), granularity, prefix).fullmatch(unquote(target)) + ) + + +def _batches_outside_layout(objects: Mapping[str, bytes], granularity: str) -> tuple[str, ...]: + def folders(body: bytes) -> frozenset[str]: + return frozenset(_folder(object_value(json.loads(line)), granularity) for line in body.splitlines()) + + return tuple( + target + for target, body in objects.items() + if len(folders(body)) != 1 + or not re.fullmatch( + re.escape(next(iter(folders(body)))) + r"batch_\d{2}-\d{2}-\d{2}_[0-9a-f]{32}\.jsonl", unquote(target) + ) + ) + + +@contextmanager +def _s3_proxy( + gateway: Gateway, + tmp_path: Path, + sink_url: str, + extra: Mapping[str, JsonValue], + settings: Mapping[str, JsonValue] | None = None, + environment: Mapping[str, str] | None = None, + workers: int = 2, + models: tuple[Mapping[str, JsonValue], ...] = (), +) -> Iterator[OwnedProxy]: + config: Final = s3_config(tmp_path, sink_url, extra, settings) + if models: + declared: Final = yaml.safe_load(config.read_text()) + config.write_text(yaml.safe_dump({**declared, "model_list": [*declared["model_list"], *models]})) + with owned_proxy_process( + gateway, tmp_path, {**FLUSH, **(environment or {})}, config=config, workers=workers + ) as owned: + yield owned + + +def _models(scenario: Scenario, provider_url: str, **key_fields: JsonValue) -> tuple[str, str, str]: + openai_model: Final = scenario.model(api_base=provider_url + "/v1", api_key="synthetic-provider-key") + anthropic_model: Final = scenario.model( + model=ANTHROPIC_MODEL, api_base=provider_url, api_key="synthetic-provider-key" + ) + return openai_model, anthropic_model, scenario.key(models=[openai_model, anthropic_model], **key_fields) + + +def _config_model(name: str, model: str, api_base: str) -> Mapping[str, JsonValue]: + return { + "model_name": name, + "litellm_params": {"model": model, "api_base": api_base, "api_key": "synthetic-provider-key"}, + } + + +def _sdk_chats(candidate: Gateway, model: str, key: str, prompts: tuple[str, ...]) -> tuple[str, ...]: + client: Final = openai.OpenAI(base_url=f"{str(candidate.client.base_url).rstrip('/')}/v1", api_key=key) + + def send(prompt: str) -> str: + reply: Final = client.chat.completions.create( + model=model, messages=[{"role": "user", "content": prompt}], extra_body={"cache": {"no-cache": True}} + ) + assert reply.choices[0].finish_reason == "stop", reply.model_dump_json() + return reply.id + + with ThreadPoolExecutor(max_workers=16) as pool: + return tuple(pool.map(send, prompts)) + + +def _surface_prompts(marker: str, per_surface: int) -> frozenset[str]: + return frozenset(f"{marker}-{surface}-{index}" for surface in SURFACES for index in range(per_surface)) + + +def _cold_storage_key(request_id: str, database_url: str | None = None) -> str: + rows: Final = eventually( + lambda: read_rows( + 'SELECT metadata FROM "LiteLLM_SpendLogs" WHERE request_id=%s', (request_id,), database_url=database_url + ), + lambda values: len(values) == 1, + seconds=60, + ) + metadata: Final = rows[0]["metadata"] + return str(object_value(json.loads(metadata) if isinstance(metadata, str) else metadata)["cold_storage_object_key"]) + + +def _update_environment(candidate: Gateway, values: Mapping[str, JsonValue]) -> None: + candidate.post( + "/config/update", + {"environment_variables": dict(values), "litellm_settings": {"success_callback": ["s3_v2"]}}, + ) + + +def _keys_on_fresh_connections(candidate: Gateway, aliases: tuple[str, ...]) -> tuple[tuple[str, str], ...]: + def generate(alias: str) -> tuple[str, str]: + with httpx.Client(base_url=candidate.client.base_url, timeout=30, trust_env=False) as fresh: + response: Final = fresh.post( + "/key/generate", + json={"key_alias": alias}, + headers={"Authorization": f"Bearer {candidate.key}", "Connection": "close"}, + ) + assert response.status_code == 200, response.text + return str(response.json()["key"]), str(response.json()["token_id"]) + + with ThreadPoolExecutor(max_workers=len(aliases)) as pool: + return tuple(pool.map(generate, aliases)) + + +def _created_key_hashes(sink: RecordingS3Sink, audit_prefix: str) -> frozenset[str]: + created: Final = ( + object_value(json.loads(body)) for target, body in sink.objects().items() if target.startswith(audit_prefix) + ) + return frozenset( + str(audit["object_id"]) + for audit in created + if audit["action"] == "created" and audit["table_name"] == "LiteLLM_VerificationToken" + ) + + +def test_s3_v2_hour_granularity_files_every_surface_under_its_hour_folder(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "s3hour" + uuid.uuid4().hex[:8] + upstream: Final = CountingUpstream() + sink: Final = RecordingS3Sink(delay_seconds=0.05) + with ( + wire_server(upstream.respond) as provider, + wire_server(sink.respond) as bucket, + _s3_proxy(gateway, tmp_path, bucket.url, HOUR) as owned, + owned.gateway.scenario() as scenario, + ): + openai_model, anthropic_model, key = _models(scenario, provider.url) + answered: Final = mixed_burst(owned.gateway, openai_model, anthropic_model, key, marker, per_surface=2) + payloads: Final = collect_payloads(sink, len(answered)) + objects: Final = sink.objects() + log: Final = owned.log.read_text() + sent: Final = _surface_prompts(marker, 2) + assert len(answered) == len(sent) and len(payloads) == len(sent), payloads + assert matched_ids(payloads, answered) == frozenset(str(payload["id"]) for payload in payloads) + assert sorted(upstream.received()) == sorted(sent) + assert len(objects) == len(sent) + assert sorted(_prompt(payload) for payload in payloads) == sorted(sent) + assert all(payload["status"] == "success" for payload in payloads), payloads + assert _outside_layout(objects, "hour") == (), "every object must sit in YYYY-MM-DD/HH/ of its start time" + assert WARNING not in log + + +@pytest.mark.parametrize( + "extra", + [ + pytest.param({}, id="missing"), + pytest.param({"s3_partition_granularity": "day"}, id="day"), + pytest.param({"s3_partition_granularity": ""}, id="empty"), + pytest.param({"s3_partition_granularity": None}, id="null"), + ], +) +def test_s3_v2_missing_day_empty_or_null_granularity_keeps_the_daily_layout( + gateway: Gateway, tmp_path: Path, extra: Mapping[str, JsonValue] +) -> None: + marker: Final = "s3day" + uuid.uuid4().hex[:8] + upstream: Final = CountingUpstream() + sink: Final = RecordingS3Sink(delay_seconds=0.05) + with ( + wire_server(upstream.respond) as provider, + wire_server(sink.respond) as bucket, + _s3_proxy(gateway, tmp_path, bucket.url, extra) as owned, + owned.gateway.scenario() as scenario, + ): + openai_model, anthropic_model, key = _models(scenario, provider.url) + answered: Final = mixed_burst(owned.gateway, openai_model, anthropic_model, key, marker, per_surface=1) + payloads: Final = collect_payloads(sink, len(answered)) + objects: Final = sink.objects() + log: Final = owned.log.read_text() + sent: Final = _surface_prompts(marker, 1) + assert len(answered) == len(sent) and len(payloads) == len(sent), payloads + assert matched_ids(payloads, answered) == frozenset(str(payload["id"]) for payload in payloads) + assert sorted(upstream.received()) == sorted(sent) + assert sorted(_prompt(payload) for payload in payloads) == sorted(sent) + assert len(objects) == len(sent) + assert _outside_layout(objects, "day") == () + assert WARNING not in log + + +@pytest.mark.parametrize( + ("extra", "environment", "shown"), + [ + pytest.param({"s3_partition_granularity": "hourly"}, {}, "'hourly'", id="unknown_word"), + pytest.param({"s3_partition_granularity": "HOUR"}, {}, "'HOUR'", id="wrong_case"), + pytest.param({"s3_partition_granularity": 1}, {}, "1", id="integer"), + pytest.param({"s3_partition_granularity": ["hour"]}, {}, "['hour']", id="list"), + pytest.param({"s3_partition_granularity": "h" * 5120}, {}, "'[base64_data truncated: 3.8KB]'", id="five_kb"), + pytest.param({}, {"S3_PARTITION_GRANULARITY": "weekly"}, "'weekly'", id="env_unknown_word"), + ], +) +def test_s3_v2_unrecognized_granularity_warns_once_per_worker_and_keeps_the_daily_layout( + gateway: Gateway, tmp_path: Path, extra: Mapping[str, JsonValue], environment: Mapping[str, str], shown: str +) -> None: + marker: Final = "s3bad" + uuid.uuid4().hex[:8] + upstream: Final = CountingUpstream() + sink: Final = RecordingS3Sink(delay_seconds=0.05) + with ( + wire_server(upstream.respond) as provider, + wire_server(sink.respond) as bucket, + _s3_proxy(gateway, tmp_path, bucket.url, extra, environment=environment) as owned, + owned.gateway.scenario() as scenario, + ): + openai_model, anthropic_model, key = _models(scenario, provider.url) + answered: Final = mixed_burst(owned.gateway, openai_model, anthropic_model, key, marker, per_surface=2) + payloads: Final = collect_payloads(sink, len(answered)) + objects: Final = sink.objects() + warning: Final = f"{WARNING}{shown} is not one of day, hour, using day" + log: Final = eventually(owned.log.read_text, lambda text: warning in text, seconds=15) + sent: Final = _surface_prompts(marker, 2) + assert len(answered) == len(sent) and len(payloads) == len(sent), payloads + assert matched_ids(payloads, answered) == frozenset(str(payload["id"]) for payload in payloads) + assert sorted(upstream.received()) == sorted(sent) + assert sorted(_prompt(payload) for payload in payloads) == sorted(sent) + assert _outside_layout(objects, "day") == () + assert 1 <= log.count(warning) <= 2, "the warning is memoized per distinct value in each of the two workers" + + +@pytest.mark.parametrize( + ("extra", "environment", "granularity"), + [ + pytest.param({}, {"S3_PARTITION_GRANULARITY": "hour"}, "hour", id="env_hour_applies"), + pytest.param({"s3_partition_granularity": "day"}, {"S3_PARTITION_GRANULARITY": "hour"}, "day", id="yaml_wins"), + ], +) +def test_s3_v2_env_granularity_applies_only_when_callback_params_leave_it_unset( + gateway: Gateway, tmp_path: Path, extra: Mapping[str, JsonValue], environment: Mapping[str, str], granularity: str +) -> None: + marker: Final = "s3env" + uuid.uuid4().hex[:8] + upstream: Final = CountingUpstream() + sink: Final = RecordingS3Sink(delay_seconds=0.05) + with ( + wire_server(upstream.respond) as provider, + wire_server(sink.respond) as bucket, + _s3_proxy(gateway, tmp_path, bucket.url, extra, environment=environment) as owned, + owned.gateway.scenario() as scenario, + ): + openai_model, _, key = _models(scenario, provider.url) + prompts: Final = tuple(f"{marker}-{index}" for index in range(8)) + returned: Final = _sdk_chats(owned.gateway, openai_model, key, prompts) + payloads: Final = collect_payloads(sink, len(prompts)) + objects: Final = sink.objects() + assert returned == prompts + assert sorted(upstream.received()) == sorted(prompts) + assert frozenset(str(payload["id"]) for payload in payloads) == frozenset(prompts) + assert _outside_layout(objects, granularity) == () + + +def test_s3_v2_hour_batch_files_group_lines_under_the_hour_folder(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "s3hbat" + uuid.uuid4().hex[:8] + upstream: Final = CountingUpstream() + sink: Final = RecordingS3Sink(delay_seconds=0.05) + with ( + wire_server(upstream.respond) as provider, + wire_server(sink.respond) as bucket, + _s3_proxy(gateway, tmp_path, bucket.url, {**HOUR, "s3_batch_file_upload": True}) as owned, + owned.gateway.scenario() as scenario, + ): + openai_model, anthropic_model, key = _models(scenario, provider.url) + answered: Final = mixed_burst(owned.gateway, openai_model, anthropic_model, key, marker, per_surface=4) + payloads: Final = collect_payloads(sink, len(answered)) + objects: Final = sink.objects() + sent: Final = _surface_prompts(marker, 4) + assert len(answered) == len(sent) and len(payloads) == len(sent), payloads + assert matched_ids(payloads, answered) == frozenset(str(payload["id"]) for payload in payloads) + assert sorted(upstream.received()) == sorted(sent) + assert sorted(_prompt(payload) for payload in payloads) == sorted(sent) + assert _batches_outside_layout(objects, "hour") == () + + +def test_s3_v2_hour_folder_sits_below_the_team_and_key_prefix(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "s3hpre" + uuid.uuid4().hex[:8] + team_alias: Final = f"alpha-{uuid.uuid4().hex[:8]}" + key_alias: Final = f"beta-{uuid.uuid4().hex[:8]}" + upstream: Final = CountingUpstream() + sink: Final = RecordingS3Sink(delay_seconds=0.05) + extra: Final = {**HOUR, "s3_use_team_prefix": True, "s3_use_key_prefix": True} + with ( + wire_server(upstream.respond) as provider, + wire_server(sink.respond) as bucket, + _s3_proxy(gateway, tmp_path, bucket.url, extra) as owned, + owned.gateway.scenario() as scenario, + ): + openai_model: Final = scenario.model(api_base=provider.url + "/v1", api_key="synthetic-provider-key") + team: Final = scenario.team(team_alias=team_alias, models=[openai_model]) + key: Final = scenario.key(team_id=team, key_alias=key_alias, models=[openai_model]) + prompts: Final = tuple(f"{marker}-{index}" for index in range(6)) + returned: Final = _sdk_chats(owned.gateway, openai_model, key, prompts) + payloads: Final = collect_payloads(sink, len(prompts)) + objects: Final = sink.objects() + assert returned == prompts + assert sorted(upstream.received()) == sorted(prompts) + assert frozenset(str(payload["id"]) for payload in payloads) == frozenset(prompts) + assert _outside_layout(objects, "hour", f"{team_alias}/{key_alias}/") == () + + +def _payload_values(payloads: tuple[dict[str, JsonValue], ...], status: str, field: str) -> frozenset[str]: + return frozenset(str(payload[field]) for payload in payloads if payload["status"] == status) + + +def test_s3_v2_hour_failure_and_rejected_requests_keep_the_hour_layout(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "s3hfail" + uuid.uuid4().hex[:8] + upstream: Final = CountingUpstream() + sink: Final = RecordingS3Sink(delay_seconds=0.05) + with ( + wire_server(upstream.respond) as provider, + wire_server(sink.respond) as bucket, + _s3_proxy(gateway, tmp_path, bucket.url, HOUR) as owned, + owned.gateway.scenario() as scenario, + ): + openai_model, _, key = _models(scenario, provider.url) + + def send(prompt: str, model: str = openai_model, caller: str = key) -> httpx.Response: + return owned.gateway.request( + "POST", + "/v1/chat/completions", + {"model": model, "messages": [{"role": "user", "content": prompt}], "cache": {"no-cache": True}}, + key=caller, + ) + + successes: Final = tuple(f"{marker}-{index}" for index in range(4)) + failures: Final = tuple(f"{marker}-{index}-fail" for index in range(3)) + with ThreadPoolExecutor(max_workers=8) as pool: + responses: Final = tuple(pool.map(send, (*successes, *failures))) + ghost: Final = send(f"{marker}-ghost", model=f"ghost-{uuid.uuid4().hex}") + unauthenticated: Final = send(f"{marker}-anon", caller="sk-not-a-real-key") + after: Final = send(f"{marker}-after") + rejected_call_ids: Final = frozenset(response.headers["x-litellm-call-id"] for response in responses[4:]) + payloads: Final = eventually( + sink.payloads, + lambda stored: ( + _payload_values(stored, "success", "id") >= frozenset((*successes, f"{marker}-after")) + and _payload_values(stored, "failure", "litellm_call_id") >= rejected_call_ids + ), + seconds=60, + ) + objects: Final = sink.objects() + assert [response.status_code for response in responses[:4]] == [200] * 4, [r.text for r in responses] + assert tuple(response.json()["id"] for response in responses[:4]) == successes + assert all(response.status_code == 401 for response in responses[4:]), [r.text for r in responses[4:]] + assert all("synthetic upstream rejection" in response.text for response in responses[4:]) + assert ghost.status_code == 403 and "key_model_access_denied" in ghost.text, ghost.text + assert unauthenticated.status_code == 401 and "error" in unauthenticated.json(), unauthenticated.text + assert after.status_code == 200 and after.json()["id"] == f"{marker}-after", after.text + assert sorted(upstream.received()) == sorted((*successes, *failures, f"{marker}-after")) + assert _payload_values(payloads, "success", "id") == frozenset((*successes, f"{marker}-after")) + assert _payload_values(payloads, "failure", "litellm_call_id") >= rejected_call_ids + assert _outside_layout(objects, "hour") == () + + +def test_s3_v2_hour_cache_hit_twins_land_one_object_each_under_the_hour_folder( + gateway: Gateway, tmp_path: Path +) -> None: + marker: Final = "s3hcache" + uuid.uuid4().hex[:8] + upstream: Final = CountingUpstream() + sink: Final = RecordingS3Sink(delay_seconds=0.05) + with ( + wire_server(upstream.respond) as provider, + wire_server(sink.respond) as bucket, + _s3_proxy(gateway, tmp_path, bucket.url, HOUR) as owned, + owned.gateway.scenario() as scenario, + ): + openai_model, anthropic_model, key = _models(scenario, provider.url) + first: Final = tuple( + call_surface(owned.gateway, surface, openai_model, anthropic_model, key, f"{marker}-{surface}", False) + for surface in ("chat", "responses") + ) + eventually(lambda: len(sink.objects()), lambda count: count >= 2, seconds=30) + repeated: Final = tuple( + call_surface(owned.gateway, surface, openai_model, anthropic_model, key, f"{marker}-{surface}", False) + for surface in ("chat", "responses") + ) + payloads: Final = collect_payloads(sink, 4) + objects: Final = sink.objects() + assert first[0][0] == f"{marker}-chat" and repeated[0][0] == first[0][0] + assert matched_ids(payloads, first + repeated) == frozenset(str(payload["id"]) for payload in payloads) + assert sorted(_prompt(payload) for payload in payloads) == sorted((f"{marker}-chat", f"{marker}-responses") * 2) + assert sorted(upstream.received()) == sorted((f"{marker}-chat", f"{marker}-responses")) + assert len(objects) == 4, list(objects) + assert sum(1 for payload in payloads if payload["cache_hit"] is True) == 2 + assert _outside_layout(objects, "hour") == () + + +def test_s3_v2_hour_cold_storage_key_names_the_uploaded_object_and_reads_back(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "s3hcold" + uuid.uuid4().hex[:8] + upstream: Final = CountingUpstream() + sink: Final = RecordingS3Sink(delay_seconds=0.05) + with ( + wire_server(upstream.respond) as provider, + wire_server(sink.respond) as bucket, + _s3_proxy(gateway, tmp_path, bucket.url, HOUR, {"cold_storage_custom_logger": "s3_v2"}) as owned, + owned.gateway.scenario() as scenario, + ): + openai_model, _, key = _models(scenario, provider.url) + prompts: Final = (f"{marker}-kept", f"{marker}-missing") + returned: Final = _sdk_chats(owned.gateway, openai_model, key, prompts) + collect_payloads(sink, len(prompts)) + objects: Final = sink.objects() + keys: Final = {prompt: _cold_storage_key(prompt) for prompt in prompts} + with sink.lock: + sink.store.pop(f"/{BUCKET}/{quote(keys[prompts[1]], safe='/')}") + kept: Final = eventually( + lambda: owned.gateway.request("GET", f"/spend/logs/ui/{prompts[0]}"), + lambda reply: reply.status_code == 200 and bool((reply.json() or {}).get("messages")), + seconds=30, + ) + missing: Final = owned.gateway.request("GET", f"/spend/logs/ui/{prompts[1]}") + assert returned == prompts + assert sorted(upstream.received()) == sorted(prompts) + assert frozenset(f"/{BUCKET}/{quote(key, safe='/')}" for key in keys.values()) == frozenset(objects) + assert _outside_layout(objects, "hour") == () + assert kept.json()["messages"] == [{"role": "user", "content": prompts[0]}], kept.text + assert prompts[0] in json.dumps(kept.json()["response"]), kept.text + assert missing.status_code == 200, missing.text + assert prompts[1] not in json.dumps(missing.json()["response"]), missing.text + + +def test_s3_v2_hour_layout_holds_when_another_logger_owns_cold_storage(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "s3hgcs" + uuid.uuid4().hex[:8] + upstream: Final = CountingUpstream() + lock: Final = threading.Lock() + puts: Final[dict[str, bytes]] = {} # mutable-ok: filled per PUT by the bucket thread under lock + + def bucket_reply(request: Request) -> Reply: + assert request.method == "PUT", request.method + with lock: + puts[unquote(request.target)] = request.body + return Reply(status=200) + + def uploaded() -> Mapping[str, bytes]: + with lock: + return dict(puts) + + with ( + wire_server(upstream.respond) as provider, + wire_server(bucket_reply) as bucket, + _s3_proxy( + gateway, tmp_path, bucket.url, {**HOUR, "s3_path": ""}, {"cold_storage_custom_logger": "gcs_bucket"} + ) as owned, + owned.gateway.scenario() as scenario, + ): + openai_model, _, key = _models(scenario, provider.url) + prompts: Final = tuple(f"{marker}-{index}" for index in range(3)) + returned: Final = _sdk_chats(owned.gateway, openai_model, key, prompts) + objects: Final = eventually(uploaded, lambda values: len(values) >= len(prompts), seconds=60) + cold_keys: Final = tuple(_cold_storage_key(prompt) for prompt in prompts) + hour_object: Final = re.compile(rf"/{BUCKET}/\d{{4}}-\d{{2}}-\d{{2}}/(\d{{2}})/time-(\d{{2}})-[^/]+\.json") + matches: Final = tuple(hour_object.fullmatch(target) for target in objects) + assert returned == prompts + assert sorted(str(object_value(json.loads(body))["id"]) for body in objects.values()) == sorted(prompts) + assert all(re.fullmatch(r"\d{4}-\d{2}-\d{2}/time-[^/]+\.json", cold_key) for cold_key in cold_keys), cold_keys + assert all(match is not None and match.group(1) == match.group(2) for match in matches), sorted(objects) + + +def test_s3_v2_hour_cold_storage_rebuilds_previous_response_id_history_from_the_hour_object( + gateway: Gateway, tmp_path: Path +) -> None: + marker: Final = "s3hsess" + uuid.uuid4().hex[:8] + upstream: Final = CountingUpstream() + histories: Final[list[str]] = [] # mutable-ok: appended per upstream request by the scripted provider thread + reads: Final[list[str]] = [] # mutable-ok: appended per sink GET by the recording sink thread + sink: Final = RecordingS3Sink(delay_seconds=0.05) + + def provider_reply(request: Request) -> Reply: + histories.append(request.body.decode()) + return upstream.respond(request) + + def bucket_reply(request: Request) -> Reply: + if request.method == "GET": + reads.append(unquote(request.target)) + return sink.respond(request) + + with ( + wire_server(provider_reply) as provider, + wire_server(bucket_reply) as bucket, + _s3_proxy(gateway, tmp_path, bucket.url, HOUR, {"cold_storage_custom_logger": "s3_v2"}) as owned, + owned.gateway.scenario() as scenario, + ): + _, anthropic_model, key = _models(scenario, provider.url) + first: Final = owned.gateway.request( + "POST", "/v1/responses", {"model": anthropic_model, "input": f"{marker}-first"}, key=key + ) + assert first.status_code == 200, first.text + rows: Final = eventually( + lambda: read_rows( + 'SELECT request_id, metadata FROM "LiteLLM_SpendLogs" WHERE model_group=%s', (anthropic_model,) + ), + lambda values: len(values) == 1, + seconds=60, + ) + metadata: Final = rows[0]["metadata"] + cold_key: Final = str( + object_value(json.loads(metadata) if isinstance(metadata, str) else metadata)["cold_storage_object_key"] + ) + eventually(sink.objects, lambda objects: f"/{BUCKET}/{quote(cold_key, safe='/')}" in objects, seconds=30) + second: Final = owned.gateway.request( + "POST", + "/v1/responses", + {"model": anthropic_model, "input": f"{marker}-second", "previous_response_id": first.json()["id"]}, + key=key, + ) + objects: Final = sink.objects() + assert second.status_code == 200, second.text + assert second.json()["id"] != first.json()["id"], second.text + assert re.fullmatch(rf"{re.escape(PREFIX)}/\d{{4}}-\d{{2}}-\d{{2}}/\d{{2}}/time-[^/]+\.json", cold_key), cold_key + assert _outside_layout(objects, "hour") == () + assert f"/{BUCKET}/{cold_key}" in reads, reads + assert len(histories) == 2, histories + assert f"{marker}-first" in histories[0] and f"{marker}-second" not in histories[0], histories[0] + assert f"{marker}-first" in histories[1] and f"{marker}-second" in histories[1], histories[1] + + +def test_s3_v2_audit_logs_follow_the_audit_params_granularity_not_the_request_logs( + gateway: Gateway, tmp_path: Path +) -> None: + marker: Final = "s3haudit" + uuid.uuid4().hex[:8] + upstream: Final = CountingUpstream() + sink: Final = RecordingS3Sink(delay_seconds=0.05) + with wire_server(upstream.respond) as provider, wire_server(sink.respond) as bucket: + settings: Final = { + "store_audit_logs": True, + "audit_log_callbacks": ["s3_v2"], + "s3_audit_callback_params": {**SINK_CREDENTIALS, "s3_endpoint_url": bucket.url, **HOUR}, + } + with ( + _s3_proxy(gateway, tmp_path, bucket.url, {}, settings) as owned, + owned.gateway.scenario() as scenario, + ): + openai_model, _, key = _models(scenario, provider.url, key_alias=marker) + returned: Final = _sdk_chats(owned.gateway, openai_model, key, (marker,)) + aliases: Final = tuple(f"{marker}-fresh{index}" for index in range(16)) + fresh_keys: Final = _keys_on_fresh_connections(owned.gateway, aliases) + audit_prefix: Final = f"/{BUCKET}/{PREFIX}/audit_logs/" + eventually( + lambda: _created_key_hashes(sink, audit_prefix), + lambda created: frozenset(token for _, token in fresh_keys) <= created, + seconds=30, + ) + owned.gateway.post("/key/delete", {"keys": [key for key, _ in fresh_keys]}) + collect_payloads(sink, 2) + objects: Final = sink.objects() + audits: Final = { + target: object_value(json.loads(body)) for target, body in objects.items() if target.startswith(audit_prefix) + } + requests: Final = {target: body for target, body in objects.items() if not target.startswith(audit_prefix)} + assert returned == (marker,) + assert upstream.received() == (marker,) + assert _outside_layout(requests, "day") == () + created: Final = tuple(audit for audit in audits.values() if audit["action"] == "created") + assert "LiteLLM_VerificationToken" in frozenset(str(audit["table_name"]) for audit in created), audits + for target, audit in audits.items(): + located: Final = re.fullmatch( + re.escape(audit_prefix) + + rf"(\d{{4}}-\d{{2}}-\d{{2}})/(\d{{2}})/(\d{{2}})-\d{{2}}-\d{{2}}_{re.escape(str(audit['id']))}\.json", + unquote(target), + ) + assert located and located[2] == located[3], (target, audit["updated_at"]) + folder: Final = datetime.fromisoformat(f"{located[1]}T{located[2]}:00:00+00:00") + updated: Final = datetime.fromisoformat(str(audit["updated_at"])) + assert timedelta(0) < folder + timedelta(hours=1) - updated <= timedelta(hours=1, minutes=1), ( + target, + audit["updated_at"], + ) + + +@pytest.mark.parametrize("level", ["key", "team"]) +def test_s3_v2_key_and_team_logging_callback_vars_cannot_change_the_proxy_hour_layout( + gateway: Gateway, tmp_path: Path, level: str +) -> None: + marker: Final = f"s3h{level}vars" + uuid.uuid4().hex[:8] + logging: Final[list[JsonValue]] = [ + {"callback_name": "s3_v2", "callback_type": "success", "callback_vars": {"s3_partition_granularity": "day"}} + ] + upstream: Final = CountingUpstream() + sink: Final = RecordingS3Sink(delay_seconds=0.05) + with ( + wire_server(upstream.respond) as provider, + wire_server(sink.respond) as bucket, + _s3_proxy(gateway, tmp_path, bucket.url, HOUR) as owned, + owned.gateway.scenario() as scenario, + ): + openai_model, _, _ = _models(scenario, provider.url) + key: Final = ( + scenario.key(models=[openai_model], metadata={"logging": logging}) + if level == "key" + else scenario.key(models=[openai_model], team_id=scenario.team(metadata={"logging": logging})) + ) + prompts: Final = tuple(f"{marker}-{index}" for index in range(8)) + returned: Final = _sdk_chats(owned.gateway, openai_model, key, prompts) + collect_payloads(sink, len(prompts)) + objects: Final = sink.objects() + assert returned == prompts + assert sorted(upstream.received()) == sorted(prompts) + assert sorted(str(object_value(json.loads(body))["id"]) for body in objects.values()) == sorted(prompts), ( + f"{level}-level s3_v2 logging must land exactly one object per request" + ) + assert _outside_layout(objects, "hour") == (), f"{level}-level callback_vars must not change the proxy granularity" + + +def test_s3_v2_admin_ui_granularity_update_moves_live_traffic_on_both_workers(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "s3hui" + uuid.uuid4().hex[:8] + upstream: Final = CountingUpstream() + sink: Final = RecordingS3Sink(delay_seconds=0.05) + with ( + scratch_database() as database_url, + wire_server(upstream.respond) as provider, + wire_server(sink.respond) as bucket, + _s3_proxy(gateway, tmp_path, bucket.url, {}, environment={"DATABASE_URL": database_url}) as owned, + owned.gateway.scenario() as scenario, + ): + openai_model, _, key = _models(scenario, provider.url) + before: Final = _sdk_chats(owned.gateway, openai_model, key, (f"{marker}-before",)) + eventually(lambda: len(sink.objects()), lambda count: count >= 1, seconds=30) + listed: Final = owned.gateway.get("/get/config/callbacks") + _update_environment(owned.gateway, {"callback": "s3_v2", "s3_partition_granularity": "hour"}) + probe_round: Final = iter(range(1000)) + + def probe() -> Mapping[str, bytes]: + round_id: Final = next(probe_round) + prompts: Final = tuple(f"{marker}-probe{round_id}-{index}" for index in range(8)) + _sdk_chats(owned.gateway, openai_model, key, prompts) + eventually( + lambda: frozenset(str(payload["id"]) for payload in sink.payloads()), + lambda landed: frozenset(prompts) <= landed, + seconds=20, + ) + return {target: body for target, body in sink.objects().items() if f"-probe{round_id}-" in target} + + eventually(probe, lambda probed: len(probed) == 8 and _outside_layout(probed, "hour") == (), seconds=60) + prompts: Final = tuple(f"{marker}-after-{index}" for index in range(16)) + returned: Final = _sdk_chats(owned.gateway, openai_model, key, prompts) + eventually( + lambda: frozenset(str(payload["id"]) for payload in sink.payloads()), + lambda landed: frozenset(prompts) <= landed, + seconds=30, + ) + after: Final = {target: body for target, body in sink.objects().items() if f"{marker}-after-" in target} + before_objects: Final = { + target: body for target, body in sink.objects().items() if f"{marker}-before" in target + } + readback: Final = owned.gateway.get("/get/config/callbacks") + s3_rows: Final = tuple(row for row in listed["callbacks"] if object_value(row)["name"] in ("s3", "s3_v2")) + assert s3_rows and all( + "S3_PARTITION_GRANULARITY" in object_value(object_value(row)["variables"]) for row in s3_rows + ), listed + after_rows: Final = tuple(row for row in readback["callbacks"] if object_value(row)["name"] in ("s3", "s3_v2")) + assert all( + object_value(object_value(row)["variables"])["S3_PARTITION_GRANULARITY"] == "hour" for row in after_rows + ), readback + assert before == (f"{marker}-before",) + assert returned == prompts + assert _outside_layout(before_objects, "day") == () + assert len(after) == len(prompts) + assert _outside_layout(after, "hour") == () + + +def test_s3_v2_granularity_toggles_mid_burst_keep_every_cold_storage_key_on_its_object( + gateway: Gateway, tmp_path: Path +) -> None: + marker: Final = "s3htog" + uuid.uuid4().hex[:8] + upstream: Final = CountingUpstream() + sink: Final = RecordingS3Sink(delay_seconds=0.05) + with ( + scratch_database() as database_url, + wire_server(upstream.respond) as provider, + wire_server(sink.respond) as bucket, + _s3_proxy( + gateway, + tmp_path, + bucket.url, + {}, + {"cold_storage_custom_logger": "s3_v2"}, + environment={"DATABASE_URL": database_url}, + ) as owned, + owned.gateway.scenario() as scenario, + ): + openai_model, _, key = _models(scenario, provider.url) + prompts: Final = tuple(f"{marker}-{index}" for index in range(32)) + with ThreadPoolExecutor(max_workers=1) as burst: + pending: Final = burst.submit(_sdk_chats, owned.gateway, openai_model, key, prompts) + for value in ("hour", "day", "hour", "day", "hour", "day"): + _update_environment(owned.gateway, {"s3_partition_granularity": value}) + returned: Final = pending.result() + collect_payloads(sink, len(prompts)) + objects: Final = sink.objects() + keys: Final = {prompt: _cold_storage_key(prompt, database_url) for prompt in prompts} + assert returned == prompts + assert sorted(upstream.received()) == sorted(prompts) + assert len(objects) == len(prompts) + assert frozenset(f"/{BUCKET}/{quote(key, safe='/')}" for key in keys.values()) == frozenset(objects), ( + "every spend log cold_storage_object_key must name the object the logger uploaded" + ) + assert all( + _object_pattern(object_value(json.loads(body)), "hour").fullmatch(unquote(target)) + or _object_pattern(object_value(json.loads(body)), "day").fullmatch(unquote(target)) + for target, body in objects.items() + ) + + +def test_s3_v2_in_flight_request_keeps_its_cold_storage_key_on_its_object_across_owner_and_granularity_switches( + gateway: Gateway, tmp_path: Path +) -> None: + marker: Final = "s3hflight" + uuid.uuid4().hex[:8] + held_prompt: Final = f"{marker}-held" + upstream: Final = CountingUpstream() + arrived: Final = threading.Event() + release: Final = threading.Event() + + def held(request: Request) -> Reply: + if held_prompt.encode() in request.body: + arrived.set() + assert release.wait(90), "held request was never released" + return upstream.respond(request) + + sink: Final = RecordingS3Sink(delay_seconds=0.05) + with ( + scratch_database() as database_url, + wire_server(held) as provider, + wire_server(sink.respond) as bucket, + _s3_proxy( + gateway, + tmp_path, + bucket.url, + {}, + {"cold_storage_custom_logger": "s3_v2"}, + environment={"DATABASE_URL": database_url}, + ) as owned, + owned.gateway.scenario() as scenario, + ): + openai_model, _, key = _models(scenario, provider.url) + with ThreadPoolExecutor(max_workers=1) as flight: + pending: Final = flight.submit(_sdk_chats, owned.gateway, openai_model, key, (held_prompt,)) + assert arrived.wait(60), "held request never reached the upstream" + owner_switch: Final = owned.gateway.request( + "POST", "/config/update", {"litellm_settings": {"cold_storage_custom_logger": "gcs_bucket"}} + ) + _update_environment(owned.gateway, HOUR) + probe_round: Final = iter(range(1000)) + + def probe() -> Mapping[str, bytes]: + round_id: Final = next(probe_round) + prompts: Final = tuple(f"{marker}-probe{round_id}-{index}" for index in range(8)) + _sdk_chats(owned.gateway, openai_model, key, prompts) + eventually( + lambda: frozenset(str(payload["id"]) for payload in sink.payloads()), + lambda landed: frozenset(prompts) <= landed, + seconds=20, + ) + return {target: body for target, body in sink.objects().items() if f"-probe{round_id}-" in target} + + eventually(probe, lambda probed: len(probed) == 8 and _outside_layout(probed, "hour") == (), seconds=60) + release.set() + returned: Final = pending.result() + eventually( + lambda: frozenset(str(payload["id"]) for payload in sink.payloads()), + lambda landed: held_prompt in landed, + seconds=30, + ) + held_objects: Final = {target: body for target, body in sink.objects().items() if held_prompt in target} + cold_key: Final = _cold_storage_key(held_prompt, database_url) + assert owner_switch.status_code == 400, owner_switch.text + assert "cold_storage_custom_logger" in owner_switch.text and "config file" in owner_switch.text, owner_switch.text + assert returned == (held_prompt,) + assert upstream.received().count(held_prompt) == 1 + assert frozenset(held_objects) == frozenset({f"/{BUCKET}/{quote(cold_key, safe='/')}"}), ( + "the in-flight request's cold_storage_object_key must name the one object the logger uploaded", + cold_key, + tuple(held_objects), + ) + assert _outside_layout(held_objects, "hour") == () + + +def test_s3_v2_cold_storage_owner_saved_through_config_update_is_not_applied_to_a_running_proxy( + gateway: Gateway, tmp_path: Path +) -> None: + marker: Final = "s3howner" + uuid.uuid4().hex[:8] + upstream: Final = CountingUpstream() + sink: Final = RecordingS3Sink() + with ( + scratch_database() as database_url, + wire_server(upstream.respond) as provider, + wire_server(sink.respond) as bucket, + _s3_proxy(gateway, tmp_path, bucket.url, HOUR, environment={"DATABASE_URL": database_url}) as owned, + owned.gateway.scenario() as scenario, + ): + openai_model, _, key = _models(scenario, provider.url) + saved: Final = owned.gateway.request( + "POST", "/config/update", {"litellm_settings": {"cold_storage_custom_logger": "s3_v2"}} + ) + prompts: Final = tuple(f"{marker}-{index}" for index in range(8)) + answered: Final = _sdk_chats(owned.gateway, openai_model, key, prompts) + landed: Final = collect_payloads(sink, len(prompts)) + rows: Final = eventually( + lambda: read_rows( + 'SELECT request_id, metadata FROM "LiteLLM_SpendLogs" WHERE request_id = ANY(%s)', + (list(answered),), + database_url=database_url, + ), + lambda values: len(values) == len(prompts), + seconds=60, + ) + objects: Final = sink.objects() + stored: Final = read_rows( + 'SELECT param_value FROM "LiteLLM_Config" WHERE param_name = %s', + ("litellm_settings",), + database_url=database_url, + ) + cold_keys: Final = { + str(row["request_id"]): object_value( + json.loads(row["metadata"]) if isinstance(row["metadata"], str) else row["metadata"] + ).get("cold_storage_object_key") + for row in rows + } + assert saved.status_code == 200, saved.text + assert [ + object_value(json.loads(row["param_value"]) if isinstance(row["param_value"], str) else row["param_value"]).get( + "cold_storage_custom_logger" + ) + for row in stored + ] == ["s3_v2"], "the owner switch must be persisted, so the unchanged live keys are not a rejected write" + assert sorted(upstream.received()) == sorted(prompts) + assert sorted(_prompt(payload) for payload in landed) == sorted(prompts) + assert cold_keys == dict.fromkeys(answered), "a DB-saved cold storage owner must not change a live request" + assert _outside_layout(objects, "hour") == () + + +def test_s3_v2_hour_postgres_outage_mid_mixed_burst_lands_every_id_exactly_once_and_recovers( + gateway: Gateway, tmp_path: Path +) -> None: + marker: Final = "s3hpg" + uuid.uuid4().hex[:8] + upstream: Final = CountingUpstream() + sink: Final = RecordingS3Sink(delay_seconds=0.05) + sent: Final = _surface_prompts(marker, 5) + openai_model: Final = f"{marker}openai" + anthropic_model: Final = f"{marker}anthropic" + with ( + scratch_database() as database_url, + database_relay(database_url, f"{marker}-".encode()) as (relay, relayed_url), + wire_server(upstream.respond) as provider, + wire_server(sink.respond) as bucket, + _s3_proxy( + gateway, + tmp_path, + bucket.url, + HOUR, + {"cold_storage_custom_logger": "s3_v2"}, + environment={"DATABASE_URL": relayed_url}, + models=( + _config_model(openai_model, "openai/gpt-4o-mini", provider.url + "/v1"), + _config_model(anthropic_model, ANTHROPIC_MODEL, provider.url), + ), + ) as owned, + owned.gateway.scenario() as scenario, + ): + key: Final = scenario.key(models=[openai_model, anthropic_model]) + warm: Final = mixed_burst(owned.gateway, openai_model, anthropic_model, key, f"{marker}warm", per_surface=2) + eventually( + lambda: frozenset(_prompt(payload) for payload in sink.payloads()), + lambda landed: _surface_prompts(f"{marker}warm", 2) <= landed, + seconds=60, + ) + relay.arm() + answered: Final = mixed_burst(owned.gateway, openai_model, anthropic_model, key, marker, per_surface=5) + assert relay.tripped.wait(90), "no spend log write reached the database during the burst" + eventually(lambda: relay.refused, lambda count: count >= 1, seconds=30) + assert relay.reconnected.wait(60), "the proxy never reconnected to the database after the outage" + burst_payloads: Final = eventually( + lambda: tuple(payload for payload in sink.payloads() if _prompt(payload) in sent), + lambda landed: frozenset(_prompt(payload) for payload in landed) == frozenset(sent), + seconds=60, + ) + recovered_prompt: Final = f"{marker}-recovered" + recovered: Final = _sdk_chats(owned.gateway, openai_model, key, (recovered_prompt,)) + recovered_key: Final = _cold_storage_key(recovered_prompt, database_url) + eventually( + lambda: frozenset(str(payload["id"]) for payload in sink.payloads()), + lambda landed: recovered_prompt in landed, + seconds=30, + ) + objects: Final = sink.objects() + uploads: Final = sink.attempts + burst: Final = burst_payloads + assert len(warm) == len(_surface_prompts(f"{marker}warm", 2)) + assert len(answered) == len(sent) == 30 + assert sorted(prompt for prompt in upstream.received() if prompt.startswith(f"{marker}-")) == sorted( + (*sent, recovered_prompt) + ) + assert matched_ids(burst, answered) == frozenset(str(payload["id"]) for payload in burst) + assert sorted(_prompt(payload) for payload in burst) == sorted(sent), "every burst id lands exactly once" + assert uploads == len(objects), "no object is uploaded twice" + assert _outside_layout(objects, "hour") == () + assert recovered == (recovered_prompt,) + assert f"/{BUCKET}/{quote(recovered_key, safe='/')}" in objects, "cold key written after recovery names its object" + + +def test_legacy_s3_callback_ignores_hour_granularity(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "s3v1hour" + uuid.uuid4().hex[:8] + upstream: Final = CountingUpstream() + sink: Final = RecordingS3Sink(delay_seconds=0.05) + with ( + wire_server(upstream.respond) as provider, + wire_server(sink.respond) as bucket, + _s3_proxy(gateway, tmp_path, bucket.url, HOUR, {"callbacks": [], "success_callback": ["s3"]}) as owned, + owned.gateway.scenario() as scenario, + ): + openai_model, _, key = _models(scenario, provider.url) + prompts: Final = tuple(f"{marker}-{index}" for index in range(3)) + returned: Final = _sdk_chats(owned.gateway, openai_model, key, prompts) + payloads: Final = collect_payloads(sink, len(prompts)) + objects: Final = sink.objects() + assert returned == prompts + assert frozenset(str(payload["id"]) for payload in payloads) == frozenset(prompts) + assert _outside_layout(objects, "day") == (), "legacy s3 keeps the daily layout, the setting is s3_v2 only" + + +def test_s3_v2_hour_sink_outage_mid_mixed_burst_lands_every_id_exactly_once(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "s3hout" + uuid.uuid4().hex[:8] + upstream: Final = CountingUpstream() + sink: Final = RecordingS3Sink(delay_seconds=0.05, fail_until=float("inf"), fail_status=503) + openai_model: Final = f"{marker}openai" + anthropic_model: Final = f"{marker}anthropic" + with ( + wire_server(upstream.respond) as provider, + wire_server(sink.respond) as bucket, + _s3_proxy( + gateway, + tmp_path, + bucket.url, + HOUR, + models=( + _config_model(openai_model, "openai/gpt-4o-mini", provider.url + "/v1"), + _config_model(anthropic_model, ANTHROPIC_MODEL, provider.url), + ), + ) as owned, + owned.gateway.scenario() as scenario, + ): + key: Final = scenario.key(models=[openai_model, anthropic_model]) + answered: Final = mixed_burst(owned.gateway, openai_model, anthropic_model, key, marker, per_surface=6) + eventually(lambda: sink.attempts, lambda attempts: attempts >= 1, seconds=30) + during: Final = owned.gateway.client.get("/health/readiness") + rejected: Final = sink.attempts + sink.fail_until = 0.0 + payloads: Final = collect_payloads(sink, len(answered), seconds=60) + objects: Final = sink.objects() + sent: Final = _surface_prompts(marker, 6) + assert len(answered) == len(sent) and len(payloads) == len(sent), payloads + assert matched_ids(payloads, answered) == frozenset(str(payload["id"]) for payload in payloads) + assert sorted(upstream.received()) == sorted(sent) + assert during.status_code == 200, during.text + assert rejected >= 1 and sink.attempts > len(objects) + assert sorted(_prompt(payload) for payload in payloads) == sorted(sent), "every burst id lands exactly once" + assert len(objects) == len(sent) + assert _outside_layout(objects, "hour") == () + + +def test_s3_v2_hour_coded_403_retries_reuse_the_same_hour_key(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "s3h403" + uuid.uuid4().hex[:8] + upstream: Final = CountingUpstream() + sink: Final = RecordingS3Sink(delay_seconds=0.05, fail_attempts=10, fail_status=403, fail_code="AccessDenied") + with ( + wire_server(upstream.respond) as provider, + wire_server(sink.respond) as bucket, + _s3_proxy(gateway, tmp_path, bucket.url, HOUR) as owned, + owned.gateway.scenario() as scenario, + ): + openai_model, _, key = _models(scenario, provider.url) + prompts: Final = tuple(f"{marker}-{index}" for index in range(16)) + returned: Final = _sdk_chats(owned.gateway, openai_model, key, prompts) + payloads: Final = collect_payloads(sink, len(prompts), seconds=60) + objects: Final = sink.objects() + attempted: Final = dict(sink.attempt_counts) + assert returned == prompts + assert sorted(str(payload["id"]) for payload in payloads) == sorted(prompts) + assert frozenset(attempted) == frozenset(objects), "a retried upload must reuse the key of its first attempt" + assert sum(attempted.values()) == len(objects) + 10 + assert _outside_layout(objects, "hour") == () + + +def test_s3_v2_hour_slow_sink_batches_never_duplicate_an_upload(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "s3hslow" + uuid.uuid4().hex[:8] + upstream: Final = CountingUpstream() + sink: Final = RecordingS3Sink(delay_seconds=1.5) + with ( + wire_server(upstream.respond) as provider, + wire_server(sink.respond) as bucket, + _s3_proxy(gateway, tmp_path, bucket.url, {**HOUR, "s3_batch_file_upload": True}) as owned, + owned.gateway.scenario() as scenario, + ): + openai_model, _, key = _models(scenario, provider.url) + prompts: Final = tuple(f"{marker}-{index}" for index in range(32)) + returned: Final = _sdk_chats(owned.gateway, openai_model, key, prompts) + + def delivered() -> int: + readiness: Final = owned.gateway.client.get("/health/readiness") + assert readiness.status_code == 200, readiness.text + return sum(len(body.splitlines()) for body in sink.objects().values()) + + eventually(delivered, lambda total: total >= len(prompts), seconds=60) + payloads: Final = sink.payloads() + objects: Final = sink.objects() + targets: Final = tuple(put.target for put in bucket.drain()) + assert returned == prompts + assert len(set(targets)) == len(targets), "the same batch object was PUT more than once" + assert sorted(str(payload["id"]) for payload in payloads) == sorted(prompts) + assert _batches_outside_layout(objects, "hour") == () + + +def _worker_processes(owned: OwnedProxy) -> tuple[int, ...]: + return tuple( + process.pid + for process in group_members(owned.process.pid) + if process.pid != owned.process.pid and "spawn_main" in " ".join(process.cmdline()) + ) + + +def test_s3_v2_hour_worker_kill_mid_burst_keeps_the_other_worker_logging(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "s3hkill" + uuid.uuid4().hex[:8] + upstream: Final = CountingUpstream() + sink: Final = RecordingS3Sink(delay_seconds=0.05) + with ( + wire_server(upstream.respond) as provider, + wire_server(sink.respond) as bucket, + _s3_proxy(gateway, tmp_path, bucket.url, HOUR) as owned, + owned.gateway.scenario() as scenario, + ): + openai_model, _, key = _models(scenario, provider.url) + workers: Final = _worker_processes(owned) + sent: Final = tuple(f"{marker}-{index}" for index in range(40)) + + def send(prompt: str) -> tuple[str, bool]: + try: + response: Final = owned.gateway.request( + "POST", + "/v1/chat/completions", + { + "model": openai_model, + "messages": [{"role": "user", "content": prompt}], + "cache": {"no-cache": True}, + }, + key=key, + ) + except httpx.HTTPError: + return prompt, False + return prompt, response.status_code == 200 and response.json()["id"] == prompt + + with ThreadPoolExecutor(max_workers=16) as pool: + futures: Final = tuple(pool.submit(send, prompt) for prompt in sent) + eventually(lambda: len(upstream.received()), lambda count: count >= 8, seconds=30) + psutil.Process(workers[0]).kill() + results: Final = tuple(future.result() for future in futures) + later: Final = tuple(f"{marker}-later-{index}" for index in range(8)) + later_results: Final = tuple(send(prompt) for prompt in later) + eventually( + lambda: frozenset(str(payload["id"]) for payload in sink.payloads()), + lambda landed: frozenset(later) <= landed, + seconds=45, + ) + payloads: Final = sink.payloads() + objects: Final = sink.objects() + assert len(workers) == 2, workers + assert all(ok for _, ok in later_results), "the surviving worker must keep serving after the kill" + landed: Final = tuple(str(payload["id"]) for payload in payloads) + assert frozenset(landed) <= frozenset((*sent, *later)), "only ids this test sent may land" + assert len(results) == len(sent), results + assert len(landed) == len(set(landed)), "no id may land twice" + assert _outside_layout(objects, "hour") == () + + +def test_s3_v2_hour_proxy_restart_mid_burst_keeps_the_layout_without_duplicates( + gateway: Gateway, tmp_path: Path +) -> None: + marker: Final = "s3hterm" + uuid.uuid4().hex[:8] + upstream: Final = CountingUpstream() + sink: Final = RecordingS3Sink(delay_seconds=0.05) + with wire_server(upstream.respond) as provider, wire_server(sink.respond) as bucket: + model_name: Final = f"integration-{marker}" + + def register(candidate: Gateway) -> str: + return str( + candidate.post( + "/model/new", + { + "model_name": model_name, + "litellm_params": { + "model": "openai/gpt-4o-mini", + "api_key": "synthetic-provider-key", + "api_base": provider.url + "/v1", + }, + "model_info": {}, + }, + )["model_info"]["id"] + ) + + def send(candidate: Gateway, key: str, prompt: str) -> tuple[str, bool]: + try: + response: Final = candidate.request( + "POST", + "/v1/chat/completions", + { + "model": model_name, + "messages": [{"role": "user", "content": prompt}], + "cache": {"no-cache": True}, + }, + key=key, + ) + except httpx.HTTPError: + return prompt, False + return prompt, response.status_code == 200 + + sent: Final = tuple(f"{marker}-{index}" for index in range(40)) + with _s3_proxy(gateway, tmp_path, bucket.url, HOUR) as first: + model_id: Final = register(first.gateway) + first_key: Final = str(first.gateway.post("/key/generate", {"models": [model_name]})["key"]) + with ThreadPoolExecutor(max_workers=16) as pool: + futures: Final = tuple(pool.submit(send, first.gateway, first_key, prompt) for prompt in sent) + eventually(lambda: len(upstream.received()), lambda count: count >= 8, seconds=30) + first.process.terminate() + results: Final = tuple(future.result() for future in futures) + first.process.wait(timeout=30) + answered: Final = frozenset(prompt for prompt, ok in results if ok) + landed_before_restart: Final = frozenset(str(payload["id"]) for payload in sink.payloads()) + with _s3_proxy(gateway, tmp_path, bucket.url, HOUR) as second: + restarted: Final = tuple(f"{marker}-restart-{index}" for index in range(8)) + second_key: Final = second.gateway.post("/key/generate", {"models": [model_name]})["key"] + restart_results: Final = tuple(send(second.gateway, str(second_key), prompt) for prompt in restarted) + eventually( + lambda: frozenset(str(payload["id"]) for payload in sink.payloads()), + lambda landed: frozenset(restarted) <= landed, + seconds=30, + ) + second.gateway.post("/model/delete", {"id": model_id}) + payloads: Final = sink.payloads() + objects: Final = sink.objects() + assert all(ok for _, ok in restart_results) + assert landed_before_restart <= answered, "a delivered object has no answered request" + landed: Final = tuple(str(payload["id"]) for payload in payloads) + assert len(landed) == len(set(landed)), "no id may land twice across the restart" + assert frozenset(restarted) <= frozenset(landed) + targets: Final = tuple(put.target for put in bucket.drain()) + assert len(set(targets)) == len(targets) + assert _outside_layout(objects, "hour") == () diff --git a/tests/unit/integrations/test_s3.py b/tests/unit/integrations/test_s3.py index fd677b9dfdf..c9a53a43d34 100644 --- a/tests/unit/integrations/test_s3.py +++ b/tests/unit/integrations/test_s3.py @@ -312,3 +312,10 @@ def test_prompts_only_payload_returns_copy_with_response_cleared(): assert stripped["messages"] == TEST_MESSAGES assert stripped is not payload assert payload == snapshot + + +def test_legacy_s3_logger_ignores_partition_granularity_and_keeps_daily_folder(): + mock_s3_client = _run_log_event({"s3_bucket_name": "b", "s3_path": "logs", "s3_partition_granularity": "hour"}) + + key = mock_s3_client.put_object.call_args.kwargs["Key"] + assert key.startswith("logs/2026-07-30/time-12-00-00-") diff --git a/tests/unit/integrations/test_s3_v2.py b/tests/unit/integrations/test_s3_v2.py index e9f5e667421..963586d9532 100644 --- a/tests/unit/integrations/test_s3_v2.py +++ b/tests/unit/integrations/test_s3_v2.py @@ -2552,6 +2552,253 @@ def test_prompts_only_toggle_is_exposed_to_admin_ui_for_both_s3_callbacks(callba assert "S3_LOG_PROMPTS_ONLY" in CustomLogger.get_callback_env_vars(callback_name) +_PARTITION_START: Final = datetime(2026, 9, 29, 14, 5, 9, 123456) +_PARTITION_ID: Final = "chatcmpl-partition" + + +def _partition_payload(response_id: str = _PARTITION_ID) -> StandardLoggingPayload: + return StandardLoggingPayload( + id=response_id, + metadata={"user_api_key_team_alias": "team-a", "user_api_key_alias": "key-a"}, + messages=[], + ) + + +def _partition_logger( + monkeypatch: pytest.MonkeyPatch, callback_params: dict[str, object], **kwargs: object +) -> S3Logger: + import litellm + + monkeypatch.setattr( + litellm, + "s3_callback_params", + {"s3_bucket_name": "test-bucket", "s3_region_name": "us-east-1", "s3_path": "logs", **callback_params}, + ) + return S3Logger( + s3_aws_access_key_id="test-key", + s3_aws_secret_access_key="test-secret", + s3_use_team_prefix=True, + s3_use_key_prefix=True, + **kwargs, + ) + + +_DAILY_KEY: Final = f"logs/team-a/key-a/2026-09-29/time-14-05-09-123456_{_PARTITION_ID}.json" +_HOURLY_KEY: Final = f"logs/team-a/key-a/2026-09-29/14/time-14-05-09-123456_{_PARTITION_ID}.json" + + +@pytest.mark.parametrize( + ("callback_params", "expected_key"), + [ + ({}, _DAILY_KEY), + ({"s3_partition_granularity": None}, _DAILY_KEY), + ({"s3_partition_granularity": "day"}, _DAILY_KEY), + ({"s3_partition_granularity": "hour"}, _HOURLY_KEY), + ], +) +def test_partition_granularity_sets_request_log_folder( + monkeypatch: pytest.MonkeyPatch, callback_params: dict[str, object], expected_key: str +) -> None: + monkeypatch.delenv("S3_PARTITION_GRANULARITY", raising=False) + logger = _partition_logger(monkeypatch, callback_params) + + element = logger.create_s3_batch_logging_element(_PARTITION_START, _partition_payload()) + + assert element is not None + assert element.s3_object_key == expected_key + + +@pytest.mark.parametrize("invalid", ["hourly", "HOUR", "1", 1, True]) +def test_invalid_partition_granularity_warns_and_keeps_daily_folder( + monkeypatch: pytest.MonkeyPatch, invalid: object +) -> None: + monkeypatch.delenv("S3_PARTITION_GRANULARITY", raising=False) + with patch("litellm.integrations.s3.verbose_logger") as mock_logger: + logger = _partition_logger(monkeypatch, {"s3_partition_granularity": invalid}) + element = logger.create_s3_batch_logging_element(_PARTITION_START, _partition_payload()) + second = logger.create_s3_batch_logging_element(_PARTITION_START, _partition_payload()) + + assert element is not None + assert second is not None + assert element.s3_object_key == second.s3_object_key == _DAILY_KEY + mock_logger.warning.assert_called_once() + assert mock_logger.warning.call_args.args[1:] == (invalid,) + + +def test_partition_granularity_reads_admin_ui_env_var_below_callback_params(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("S3_PARTITION_GRANULARITY", "hour") + + from_env = _partition_logger(monkeypatch, {}).create_s3_batch_logging_element( + _PARTITION_START, _partition_payload() + ) + from_params = _partition_logger(monkeypatch, {"s3_partition_granularity": "day"}).create_s3_batch_logging_element( + _PARTITION_START, _partition_payload() + ) + + assert from_env is not None and from_env.s3_object_key == _HOURLY_KEY + assert from_params is not None and from_params.s3_object_key == _DAILY_KEY + + +def test_partition_granularity_constructor_argument_and_os_environ_reference(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("S3_PARTITION_GRANULARITY", raising=False) + monkeypatch.setenv("MY_S3_PARTITION", "hour") + + from_ctor = _partition_logger(monkeypatch, {}, s3_partition_granularity="hour") + from_secret = _partition_logger(monkeypatch, {"s3_partition_granularity": "os.environ/MY_S3_PARTITION"}) + + for logger in (from_ctor, from_secret): + element = logger.create_s3_batch_logging_element(_PARTITION_START, _partition_payload()) + assert element is not None and element.s3_object_key == _HOURLY_KEY + + +def test_hourly_partition_long_key_keeps_hour_folder_within_s3_limit(monkeypatch: pytest.MonkeyPatch) -> None: + from litellm.constants import MAX_S3_OBJECT_KEY_BYTES + + monkeypatch.delenv("S3_PARTITION_GRANULARITY", raising=False) + logger = _partition_logger(monkeypatch, {"s3_partition_granularity": "hour", "s3_path": "p" * 1100}) + + element = logger.create_s3_batch_logging_element(_PARTITION_START, _partition_payload("r" * 600)) + + assert element is not None + assert len(element.s3_object_key.encode("utf-8")) <= MAX_S3_OBJECT_KEY_BYTES + assert re.search(r"/2026-09-29/14/[0-9a-f]{64}\.json$", element.s3_object_key) + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("granularity", "hour_folder"), [("hour", True), ("day", False), (None, False)]) +async def test_audit_log_key_follows_audit_callback_params_partition_granularity( + monkeypatch: pytest.MonkeyPatch, granularity: str | None, hour_folder: bool +) -> None: + monkeypatch.delenv("S3_PARTITION_GRANULARITY", raising=False) + logger = S3Logger( + s3_callback_params_override={ + "s3_bucket_name": "audit-bucket", + "s3_path": "audit", + "s3_partition_granularity": granularity, + } + ) + + await logger.async_log_audit_log_event({"id": "audit-1"}) + + (element,) = logger.log_queue + match = re.fullmatch( + r"audit/audit_logs/\d{4}-\d{2}-\d{2}/(?:(\d{2})/)?(\d{2})-\d{2}-\d{2}_audit-1\.json", element.s3_object_key + ) + assert match is not None, element.s3_object_key + assert (match.group(1) is not None) is hour_folder + if hour_folder: + assert match.group(1) == match.group(2) + + +@pytest.mark.asyncio +async def test_hourly_batch_file_upload_writes_one_file_per_hour_folder(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("S3_PARTITION_GRANULARITY", raising=False) + logger = _partition_logger(monkeypatch, {"s3_partition_granularity": "hour"}, s3_batch_file_upload=True) + put = _RecordingPut() + logger.async_httpx_client = AsyncMock() + logger.async_httpx_client.put = put + before = logger.create_s3_batch_logging_element(datetime(2026, 9, 29, 13, 59, 59), _partition_payload("before")) + after = logger.create_s3_batch_logging_element(datetime(2026, 9, 29, 14, 0, 1), _partition_payload("after")) + assert before is not None and after is not None + logger.log_queue = [before, after] + + await logger.async_send_batch() + + by_folder = { + re.sub(r"/batch_\d{2}-\d{2}-\d{2}_[0-9a-f]{32}\.jsonl$", "", url.split(".com/", 1)[-1]): data + for url, data, _headers in put.calls + } + assert sorted(by_folder) == ["logs/team-a/key-a/2026-09-29/13", "logs/team-a/key-a/2026-09-29/14"] + assert [json.loads(line)["id"] for line in (by_folder["logs/team-a/key-a/2026-09-29/13"] or "").splitlines()] == [ + "before" + ] + assert [json.loads(line)["id"] for line in (by_folder["logs/team-a/key-a/2026-09-29/14"] or "").splitlines()] == [ + "after" + ] + + +@pytest.mark.parametrize("granularity", [None, "day", "hour"]) +def test_cold_storage_object_key_matches_the_uploaded_request_log_key( + monkeypatch: pytest.MonkeyPatch, granularity: str | None +) -> None: + import litellm + from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup + + monkeypatch.delenv("S3_PARTITION_GRANULARITY", raising=False) + monkeypatch.setattr( + litellm, + "s3_callback_params", + {"s3_bucket_name": "test-bucket", "s3_path": "coldlogs", "s3_partition_granularity": granularity}, + ) + monkeypatch.setattr(litellm, "cold_storage_custom_logger", "s3_v2") + logger = S3Logger() + uploaded = logger.create_s3_batch_logging_element( + _PARTITION_START, StandardLoggingPayload(id=_PARTITION_ID, metadata={}, messages=[]) + ) + + monkeypatch.setattr(litellm, "callbacks", [logger]) + cold_key = StandardLoggingPayloadSetup._generate_cold_storage_object_key( + start_time=_PARTITION_START, response_id=_PARTITION_ID + ) + + assert uploaded is not None + assert cold_key == uploaded.s3_object_key + assert ("/2026-09-29/14/" in cold_key) is (granularity == "hour") + + +def test_cold_storage_key_matches_upload_when_env_var_changes_mid_request(monkeypatch: pytest.MonkeyPatch) -> None: + import litellm + from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup + + monkeypatch.delenv("S3_PARTITION_GRANULARITY", raising=False) + monkeypatch.setattr(litellm, "s3_callback_params", {"s3_bucket_name": "test-bucket", "s3_path": "coldlogs"}) + monkeypatch.setattr(litellm, "cold_storage_custom_logger", "s3_v2") + logger = S3Logger() + monkeypatch.setattr(litellm, "callbacks", [logger]) + + cold_key = StandardLoggingPayloadSetup._generate_cold_storage_object_key( + start_time=_PARTITION_START, response_id=_PARTITION_ID + ) + monkeypatch.setenv("S3_PARTITION_GRANULARITY", "hour") + uploaded = logger.create_s3_batch_logging_element( + _PARTITION_START, + StandardLoggingPayload(id=_PARTITION_ID, metadata={"cold_storage_object_key": cold_key}, messages=[]), + ) + + assert uploaded is not None + assert cold_key == uploaded.s3_object_key == f"coldlogs/2026-09-29/time-14-05-09-123456_{_PARTITION_ID}.json" + + +def test_hour_upload_ignores_a_cold_storage_key_owned_by_another_logger(monkeypatch: pytest.MonkeyPatch) -> None: + import litellm + from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup + + monkeypatch.delenv("S3_PARTITION_GRANULARITY", raising=False) + monkeypatch.setattr( + litellm, "s3_callback_params", {"s3_bucket_name": "test-bucket", "s3_partition_granularity": "hour"} + ) + monkeypatch.setattr(litellm, "cold_storage_custom_logger", "gcs_bucket") + logger = S3Logger() + cold_key = StandardLoggingPayloadSetup._generate_cold_storage_object_key( + start_time=_PARTITION_START, response_id=_PARTITION_ID + ) + uploaded = logger.create_s3_batch_logging_element( + _PARTITION_START, + StandardLoggingPayload(id=_PARTITION_ID, metadata={"cold_storage_object_key": cold_key}, messages=[]), + ) + + assert cold_key == f"2026-09-29/time-14-05-09-123456_{_PARTITION_ID}.json" + assert uploaded is not None + assert uploaded.s3_object_key == f"2026-09-29/14/time-14-05-09-123456_{_PARTITION_ID}.json" + + +@pytest.mark.parametrize("callback_name", ["s3", "s3_v2"]) +def test_partition_granularity_is_exposed_to_admin_ui(callback_name: str) -> None: + from litellm.integrations.custom_logger import CustomLogger + + assert "S3_PARTITION_GRANULARITY" in CustomLogger.get_callback_env_vars(callback_name) + + def _element(payload: dict[str, object], key_suffix: str) -> s3BatchLoggingElement: return s3BatchLoggingElement( s3_object_key=f"2025-09-14/test-{key_suffix}.json", diff --git a/tests/unit/litellm_core_utils/test_litellm_logging.py b/tests/unit/litellm_core_utils/test_litellm_logging.py index 761fa38e73f..e07ffe00d4c 100644 --- a/tests/unit/litellm_core_utils/test_litellm_logging.py +++ b/tests/unit/litellm_core_utils/test_litellm_logging.py @@ -3227,6 +3227,7 @@ async def test_e2e_generate_cold_storage_object_key_successful(): prefix="", # No prefix for cold storage start_time=start_time, s3_file_name="time-10-30-45-123456_chatcmpl-test-12345", + partition_granularity="day", ) # Verify the result @@ -3276,6 +3277,7 @@ async def test_e2e_generate_cold_storage_object_key_with_custom_logger_s3_path() prefix="", start_time=start_time, s3_file_name="time-10-30-45-123456_chatcmpl-test-12345", + partition_granularity="day", ) # Verify the result @@ -3320,6 +3322,7 @@ async def test_e2e_generate_cold_storage_object_key_with_logger_no_s3_path(): prefix="", start_time=start_time, s3_file_name="time-10-30-45-123456_chatcmpl-test-12345", + partition_granularity="day", ) # Verify the result diff --git a/ui/litellm-dashboard/src/components/settings.test.tsx b/ui/litellm-dashboard/src/components/settings.test.tsx index 4ba5dd23fd1..9b67657dbca 100644 --- a/ui/litellm-dashboard/src/components/settings.test.tsx +++ b/ui/litellm-dashboard/src/components/settings.test.tsx @@ -313,6 +313,7 @@ describe("Settings", () => { "AWS_SECRET_ACCESS_KEY", "AWS_REGION_NAME", "S3_LOG_PROMPTS_ONLY", + "S3_PARTITION_GRANULARITY", ], ui_callback_name: "s3 Bucket (AWS)", }, @@ -326,6 +327,12 @@ describe("Settings", () => { dynamic_params: { s3_bucket_name: { type: "text", ui_name: "S3 Bucket Name", required: false }, s3_log_prompts_only: { type: "boolean", ui_name: "Log Prompts Only", required: false }, + s3_partition_granularity: { + type: "select", + ui_name: "Folder Partitioning", + options: ["day", "hour"], + required: false, + }, }, }, ]); @@ -409,6 +416,46 @@ describe("Settings", () => { }); }); + it("should show the saved s3_v2 folder partitioning and post the newly selected value", async () => { + mockS3Callback({ S3_LOG_PROMPTS_ONLY: null, S3_PARTITION_GRANULARITY: "hour" }, "s3_v2"); + const user = await openS3EditModal("s3_v2"); + + const dialog = screen.getByRole("dialog"); + const partitioning = await within(dialog).findByRole("combobox", { name: "Folder Partitioning" }); + expect(partitioning).toHaveTextContent("hour"); + + await user.click(partitioning); + await user.click(await screen.findByRole("option", { name: "day" })); + await user.click(within(dialog).getByRole("button", { name: "Save Changes" })); + + await waitFor(() => { + expect(vi.mocked(setCallbacksCall)).toHaveBeenCalledWith( + "token", + expect.objectContaining({ + environment_variables: expect.objectContaining({ callback: "s3_v2", s3_partition_granularity: "day" }), + litellm_settings: { success_callback: ["s3_v2"] }, + }), + ); + }); + }); + + it("should not offer folder partitioning for the legacy s3 callback, which cannot honour it", async () => { + mockS3Callback({ S3_LOG_PROMPTS_ONLY: null, S3_PARTITION_GRANULARITY: null }); + const user = await openS3EditModal(); + + const dialog = screen.getByRole("dialog"); + await within(dialog).findByRole("switch", { name: "Log Prompts Only" }); + expect(within(dialog).queryByRole("combobox", { name: "Folder Partitioning" })).not.toBeInTheDocument(); + + await user.click(within(dialog).getByRole("button", { name: "Save Changes" })); + await waitFor(() => { + expect(vi.mocked(setCallbacksCall)).toHaveBeenCalledTimes(1); + }); + const [, payload] = vi.mocked(setCallbacksCall).mock.calls[0]; + expect(payload.environment_variables).not.toHaveProperty("s3_partition_granularity"); + expect(payload.environment_variables).not.toHaveProperty("S3_PARTITION_GRANULARITY"); + }); + it("should send the typed webhook url for an alert type when the alerting tab is saved", async () => { const user = userEvent.setup(); render(); diff --git a/ui/litellm-dashboard/src/components/settings.tsx b/ui/litellm-dashboard/src/components/settings.tsx index 9247f22ec28..e376d858df8 100644 --- a/ui/litellm-dashboard/src/components/settings.tsx +++ b/ui/litellm-dashboard/src/components/settings.tsx @@ -238,6 +238,7 @@ export const CallbackSelector: React.FC = ({ }; const CALLBACK_CONFIG_ALIASES: Record = { s3_v2: "s3" }; +const CALLBACK_UNSUPPORTED_PARAMS: Record = { s3: ["s3_partition_granularity"] }; interface DynamicParamConfig { type?: string; @@ -274,7 +275,8 @@ const getDynamicParamsForCallback = ( const callbackConfig = findCallbackConfig(callbackConfigs, callbackName); if (callbackConfig?.dynamic_params) { - return Object.keys(callbackConfig.dynamic_params); + const unsupportedParams = CALLBACK_UNSUPPORTED_PARAMS[callbackName] ?? []; + return Object.keys(callbackConfig.dynamic_params).filter((param) => !unsupportedParams.includes(param)); } return fallbackVariables ? Object.keys(fallbackVariables) : [];