diff --git a/gateway/routes/allowlist.py b/gateway/routes/allowlist.py index 92b73867e67..3733072a948 100644 --- a/gateway/routes/allowlist.py +++ b/gateway/routes/allowlist.py @@ -3,7 +3,8 @@ The gateway exposes the LLM data-plane surface: chat/completions, embeddings, audio, batches, files, fine-tuning, rerank, ocr, rag, video, search, image, responses, vector stores, passthrough providers, realtime websockets, MCP -tool-call endpoints, and operational endpoints (/health, /metrics). +tool-call endpoints, and operational endpoints (/health, /metrics, and the +/debug/memory/summary read of the serving worker's RSS). Any path not listed here is dropped from the gateway process so management/UI endpoints don't ride on the same pods. @@ -121,6 +122,7 @@ GATEWAY_EXACT_PATHS: frozenset[str] = frozenset( "/docs/oauth2-redirect", "/redoc", "/test", + "/debug/memory/summary", } ) diff --git a/helm/litellm/templates/ingress.yaml b/helm/litellm/templates/ingress.yaml index 732564b280f..d42558b9396 100644 --- a/helm/litellm/templates/ingress.yaml +++ b/helm/litellm/templates/ingress.yaml @@ -89,7 +89,7 @@ at "/" Prefix would swallow the whole backend management API) instead of adding to it. */}} -{{- $builtinPathKeys := list "/test|Exact" "/|Prefix" -}} +{{- $builtinPathKeys := list "/test|Exact" "/debug/memory/summary|Exact" "/|Prefix" -}} apiVersion: networking.k8s.io/v1 kind: Ingress metadata: @@ -129,6 +129,8 @@ spec: # --- Gateway data plane --- # Exact /test only (see the $gatewayPrefixes comment above); # /test/* MCP management endpoints fall to the backend catch-all. + # Exact /debug/memory/summary reads a serving worker's RSS (the e2e memory + # gate); the rest of /debug/* stays on the backend. - path: /test pathType: Exact backend: @@ -136,6 +138,13 @@ spec: name: {{ $gatewayName }} port: number: {{ $gatewayPort }} + - path: /debug/memory/summary + pathType: Exact + backend: + service: + name: {{ $gatewayName }} + port: + number: {{ $gatewayPort }} {{- range $gatewayPrefixes }} {{- $pathType := include "litellm.ingress.pathType" (dict "controller" $controller "path" . "pathType" "Prefix") }} {{- $builtinPathKeys = append $builtinPathKeys (printf "%s|%s" . $pathType) }} diff --git a/helm/litellm/tests/ingress_controller_tests.yaml b/helm/litellm/tests/ingress_controller_tests.yaml index 40790ba674a..aa30db3c9c1 100644 --- a/helm/litellm/tests/ingress_controller_tests.yaml +++ b/helm/litellm/tests/ingress_controller_tests.yaml @@ -97,6 +97,16 @@ tests: name: RELEASE-NAME-litellm-gateway port: number: 4000 + - contains: + path: spec.rules[0].http.paths + content: + path: /debug/memory/summary + pathType: Exact + backend: + service: + name: RELEASE-NAME-litellm-gateway + port: + number: 4000 - equal: path: spec.rules[0].http.paths[-1] value: diff --git a/helm/litellm/tests/ingress_extra_paths_tests.yaml b/helm/litellm/tests/ingress_extra_paths_tests.yaml index fc7d5943278..1305af15ae2 100644 --- a/helm/litellm/tests/ingress_extra_paths_tests.yaml +++ b/helm/litellm/tests/ingress_extra_paths_tests.yaml @@ -288,6 +288,17 @@ tests: - failedTemplate: errorMessage: "ingress.extraPaths[0]: path /test with pathType Exact is already routed by this chart, and a duplicate would take it over rather than add to it" + - it: rejects an entry that would take over the exact /debug/memory/summary route + set: + ingress.enabled: true + ingress.extraPaths: + - path: /debug/memory/summary + pathType: Exact + service: backend + asserts: + - failedTemplate: + errorMessage: "ingress.extraPaths[0]: path /debug/memory/summary with pathType Exact is already routed by this chart, and a duplicate would take it over rather than add to it" + - it: allows a built-in path under a different pathType, which is a distinct rule set: ingress.enabled: true diff --git a/litellm/proxy/common_utils/debug_utils.py b/litellm/proxy/common_utils/debug_utils.py index 1d4024bb84e..dc329e55e31 100644 --- a/litellm/proxy/common_utils/debug_utils.py +++ b/litellm/proxy/common_utils/debug_utils.py @@ -3,6 +3,7 @@ import asyncio import gc import json import os +import socket import sys import tracemalloc from collections import Counter @@ -237,6 +238,80 @@ def _process_memory_usage(process: _ProcessHandle) -> _ProcessMemoryUsage: ) +PROC_STATM_PATH: Final = "/proc/self/statm" +PROC_MEMINFO_PATH: Final = "/proc/meminfo" +PSUTIL_MISSING_ERROR: Final = "Install psutil for memory monitoring: pip install psutil" + + +class _ProcMemoryInfo(NamedTuple): + rss: int + vms: int + + +class _ProcFilesystemProcess: + """Memory of the running process read from the Linux proc filesystem, for images without psutil.""" + + def __init__( + self, + statm_path: str = PROC_STATM_PATH, + meminfo_path: str = PROC_MEMINFO_PATH, + page_size: int | None = None, + ) -> None: + self._statm_path: Final = statm_path + self._meminfo_path: Final = meminfo_path + self._page_size: Final = os.sysconf("SC_PAGE_SIZE") if page_size is None else page_size + + def memory_info(self) -> _ProcMemoryInfo: + with open(self._statm_path, encoding="ascii") as statm: + size_pages, resident_pages = statm.read().split()[:2] + return _ProcMemoryInfo(rss=int(resident_pages) * self._page_size, vms=int(size_pages) * self._page_size) + + def memory_percent(self) -> float: + with open(self._meminfo_path, encoding="ascii") as meminfo: + total_kilobytes: Final = next(int(line.split()[1]) for line in meminfo if line.startswith("MemTotal:")) + return self.memory_info().rss / (total_kilobytes * 1024) * 100 + + +def _process_handle() -> _ProcessHandle | None: + try: + import psutil + except ImportError: + return _ProcFilesystemProcess() if os.path.exists(PROC_STATM_PATH) else None + return psutil.Process() + + +def _health_status(memory_percent: float) -> str: + if memory_percent > 80: + return "critical" + if memory_percent > 60: + return "warning" + return "healthy" + + +class _SummaryProcessMemory(TypedDict, total=False): + summary: ReadOnly[str] + ram_usage_mb: ReadOnly[float] + system_memory_percent: ReadOnly[float] + error: ReadOnly[str] + + +def _summary_process_memory(process: _ProcessHandle | None) -> tuple[_SummaryProcessMemory, str]: + if process is None: + missing: Final[_SummaryProcessMemory] = {"error": PSUTIL_MISSING_ERROR} + return missing, "healthy" + try: + usage: Final = _process_memory_usage(process) + except Exception as e: + unreadable: Final[_SummaryProcessMemory] = {"error": str(e)} + return unreadable, "healthy" + memory: Final[_SummaryProcessMemory] = { + "summary": f"{usage.resident_megabytes:.1f} MB ({usage.percent:.1f}% of system memory)", + "ram_usage_mb": round(usage.resident_megabytes, 2), + "system_memory_percent": round(usage.percent, 2), + } + return memory, _health_status(usage.percent) + + @router.get("/debug/memory/summary", include_in_schema=False) async def get_memory_summary( _: UserAPIKeyAuth = Depends(user_api_key_auth), @@ -246,6 +321,7 @@ async def get_memory_summary( Returns: - worker_pid: Process ID + - hostname: Host (the pod on Kubernetes) the worker runs on - status: Overall health based on memory usage - memory: Process memory usage and RAM info - caches: Cache item counts and descriptions @@ -263,35 +339,7 @@ async def get_memory_summary( user_api_key_cache, ) - # Get process memory info - process_memory = {} - health_status = "healthy" - - try: - import psutil - - usage: Final = _process_memory_usage(psutil.Process()) - memory_mb: Final = usage.resident_megabytes - memory_percent: Final = usage.percent - - process_memory = { - "summary": f"{memory_mb:.1f} MB ({memory_percent:.1f}% of system memory)", - "ram_usage_mb": round(memory_mb, 2), - "system_memory_percent": round(memory_percent, 2), - } - - # Check memory health status - if memory_percent > 80: - health_status = "critical" - elif memory_percent > 60: - health_status = "warning" - else: - health_status = "healthy" - - except ImportError: - process_memory["error"] = "Install psutil for memory monitoring: pip install psutil" - except Exception as e: - process_memory["error"] = str(e) + process_memory, health_status = _summary_process_memory(_process_handle()) # Get cache information caches: Final[dict[str, object]] = {} @@ -347,6 +395,7 @@ async def get_memory_summary( return { "worker_pid": os.getpid(), + "hostname": socket.gethostname(), "status": health_status, "memory": process_memory, "caches": { diff --git a/tests/e2e/CLAUDE.md b/tests/e2e/CLAUDE.md index 8be2e7b4ce2..867afecc9da 100644 --- a/tests/e2e/CLAUDE.md +++ b/tests/e2e/CLAUDE.md @@ -17,7 +17,7 @@ Each subdirectory under `tests/e2e/` is one suite, scoped to an endpoint family - `mcp/` - the MCP server surface over api_key auth against the real Datadog remote MCP server (see "MCP suite: real Datadog only" below); plus the gateway-managed OAuth (authorization_code) path exercised through `/chat/completions`, the one behavior Datadog's static-header auth cannot reach, seeding the per-user upstream token via the interactive authorize dance driven with the mcp SDK's own OAuth client (headless-browser consent from a saved session) and asserting the completion lists and executes the server's tools with the stored per-user token - `logging/` - logging-integration delivery (datadog and friends) - `security/` - secret handling and log-leak protection -- `router/` - routing and reliability behavior (fallbacks, cooldowns) +- `router/` - routing and reliability behavior (fallbacks, cooldowns) plus the memory regression test (`test_reliability_memory_e2e.py`: a few hundred failing requests with retries and fallbacks must not grow proxy RSS past a fixed budget nor store a request snapshot past a fixed size, the release-gate check for the v1.100.0 retry-breadcrumb leak) - `load/` - performance-category tests, kept OUT of the main suite: throughput/load SLO tests are a different testing category from functional e2e (variance-driven, historically flaky) and live outside this suite until re-implemented as their own pipeline (LIT-5163); do not add a live load test that runs in the default collection. What lives here: the weekly session-anomaly test (`test_weekly_session_anomaly_e2e.py`, Claude Code-shaped multi-turn sessions against real providers with ceilings on error rate, cache read/write, turn time, and spend; marked `weekly` and deselected unless `E2E_WEEKLY_ANOMALY` is set, driven by `.github/workflows/weekly_load_anomaly.yml`), the Redis chaos test (`test_redis_chaos_e2e.py`, locust load against mock deployments split round robin over `/chat/completions` and `/v1/messages`, one endpoint per simulated user, with `CLIENT PAUSE ALL` on the proxy's Redis mid-run to simulate it being down outright, asserting zero failed requests on every endpoint, budgeting RSS and CPU-per-request as ratios against the same run's healthy phase, and holding p50/p90/p99 latency and log-bytes-per-request to flat ceilings (a ratio cannot bound those two: an open breaker skips Redis instead of waiting on it, so the chaos phase can measure cheaper than baseline while still being far slower than a user should see); needs a proxy booted from `gateway/redis_chaos_ci_config.yml` on the same host with `E2E_PROXY_PID` and `E2E_PROXY_LOG` set, marked `redis_chaos`, deselected unless `E2E_REDIS_CHAOS` is set and excluded from the per-PR selector like the rest of `load/`, driven by `.github/workflows/test-e2e-redis-chaos.yml` and by the Buildkite `e2e-redis-chaos` step in project-releaser, which runs the proxy, Postgres and Valkey co-located with pytest in one pod and sets the opt-in), and markerless harness unit tests for the locust, process-usage, and session-anomaly aggregation logic - `other/` - the holding-pen suite for the `other.*` registry cluster with no home of its own yet: the master-key auth gate and the process-lifecycle health probes (liveness, public readiness, authenticated readiness diagnostics). Promote a cluster out once it is large/stable enough for its own suite - `gateway/` - proxy configuration only (`litellm-config.yml`); no tests @@ -163,7 +163,7 @@ reliability... behavior : fallback | retry | cooldown | timeout | routing | cache | circuit_breaker | perf variant : 5xx | context_window | content_policy | 429 | timeout simple_shuffle | usage_based | latency_based | cost_based | least_busy - latency | throughput | session_anomaly (perf only; SLO/threshold assertion, not binary) + latency | throughput | session_anomaly | memory (perf only; SLO/threshold assertion, not binary) assertion : routes_to_fallback | succeeds_within_retries | picks_under_tpm | returns_cached | trips_then_recovers | under_slo e.g. reliability.fallback.context_window.routes_to_fallback exercised_on=[chat_completions] diff --git a/tests/e2e/coverage_registry/reliability.yaml b/tests/e2e/coverage_registry/reliability.yaml index cfbedf3d236..e95d27bab84 100644 --- a/tests/e2e/coverage_registry/reliability.yaml +++ b/tests/e2e/coverage_registry/reliability.yaml @@ -35,4 +35,5 @@ - {id: reliability.timeout.request_timeout.exceeds_deadline, module: reliability, tier: P1, behavior: timeout, variant: request_timeout, assertions: [exceeds_deadline], exercised_on: [chat_completions, messages], source: "litellm/router.py:545-551", rationale: "Per-request timeout raises Timeout"} - {id: reliability.timeout.stream_timeout.exceeds_deadline, module: reliability, tier: P1, behavior: timeout, variant: stream_timeout, assertions: [exceeds_deadline], exercised_on: [chat_completions], source: "litellm/router.py:551", rationale: "Streaming chunk-delivery timeout"} - {id: reliability.perf.throughput.under_slo, module: reliability, tier: P1, behavior: perf, variant: throughput, assertions: [under_slo], exercised_on: [chat_completions, messages], source: grammar, rationale: "Throughput SLO under load"} +- {id: reliability.perf.memory.under_slo, module: reliability, tier: P1, behavior: perf, variant: memory, assertions: [under_slo], exercised_on: [chat_completions], source: grammar, rationale: "Proxy RSS and the stored request snapshot stay within fixed budgets across a few hundred failing requests with retries and fallbacks, the v1.100.0 retry-breadcrumb leak shape (MAT-335)"} - {id: reliability.perf.session_anomaly.under_slo, module: reliability, tier: P1, behavior: perf, variant: session_anomaly, assertions: [under_slo], exercised_on: [messages], source: grammar, rationale: "Weekly Claude Code-shaped multi-turn session load against real providers; ceilings on error rate, warm-turn cache read/write, p95 turn time, and gateway-recorded spend (LIT-4562)"} diff --git a/tests/e2e/e2e_config.py b/tests/e2e/e2e_config.py index 7344a2b6cec..e15a0cd0f8f 100644 --- a/tests/e2e/e2e_config.py +++ b/tests/e2e/e2e_config.py @@ -160,6 +160,14 @@ ANOMALY_MAX_KEY_SPEND_USD = float( ANOMALY_SPEND_SETTLE_SECONDS = float( os.environ.get("E2E_ANOMALY_SPEND_SETTLE_SECONDS", "75") ) +MEMORY_REQUESTS_PER_PHASE = int(os.environ.get("E2E_MEMORY_REQUESTS_PER_PHASE", "300")) +MEMORY_RETRIES_PER_REQUEST = int(os.environ.get("E2E_MEMORY_RETRIES_PER_REQUEST", "2")) +MEMORY_TRANSCRIPT_TURNS = int(os.environ.get("E2E_MEMORY_TRANSCRIPT_TURNS", "40")) +MEMORY_CONCURRENCY = int(os.environ.get("E2E_MEMORY_CONCURRENCY", "4")) +MEMORY_RSS_SETTLE_SAMPLES = int(os.environ.get("E2E_MEMORY_RSS_SETTLE_SAMPLES", "15")) +MEMORY_RSS_SAMPLE_INTERVAL_SECONDS = float(os.environ.get("E2E_MEMORY_RSS_SAMPLE_INTERVAL_SECONDS", "1")) +MEMORY_RSS_BUDGET_MB = float(os.environ.get("E2E_MEMORY_RSS_BUDGET_MB", "48")) +MEMORY_STORED_REQUEST_BUDGET_KB = float(os.environ.get("E2E_MEMORY_STORED_REQUEST_BUDGET_KB", "64")) def ws_base_url() -> str: diff --git a/tests/e2e/models.py b/tests/e2e/models.py index 2a728d0ef40..f362d4cc6e5 100644 --- a/tests/e2e/models.py +++ b/tests/e2e/models.py @@ -11,7 +11,16 @@ from datetime import datetime from typing import Final, Literal from e2e_http import PartialBody -from pydantic import AliasChoices, BaseModel, ConfigDict, Field, RootModel, model_serializer, model_validator +from pydantic import ( + AliasChoices, + BaseModel, + ConfigDict, + Field, + JsonValue, + RootModel, + model_serializer, + model_validator, +) # ---------- keys ---------- @@ -695,6 +704,7 @@ class SpendLogRow(BaseModel): total_tokens: int | None = None request_tags: list[str] | None = None metadata: SpendLogMetadata | None = None + proxy_server_request: JsonValue = None class SpendLogs(RootModel[list[SpendLogRow]]): @@ -926,6 +936,7 @@ class LiteLLMParamsBody(BaseModel): timeout: float | None = None tpm: int | None = None weight: int | None = None + cooldown_time: float | None = None order: int | None = None @@ -1261,6 +1272,19 @@ class TagListResponse(RootModel[list[TagListEntry]]): # ---------- health / lifecycle ---------- +class ProcessMemory(BaseModel): + ram_usage_mb: float | None = None + system_memory_percent: float | None = None + error: str | None = None + + +class MemorySummaryResponse(BaseModel): + worker_pid: int + hostname: str | None = None + status: str + memory: ProcessMemory + + class ReadinessResponse(BaseModel): """GET /health/readiness (public probe). The low-detail payload a load balancer sees: `status` plus the resolved DB state (`connected`, diff --git a/tests/e2e/proxy_client.py b/tests/e2e/proxy_client.py index 1bac5116a9d..2e0a23ce58b 100644 --- a/tests/e2e/proxy_client.py +++ b/tests/e2e/proxy_client.py @@ -63,6 +63,7 @@ from models import ( ModelNewBody, ModelNewResponse, ModelsListParams, + MemorySummaryResponse, ModelsListResponse, ModelUpdateBody, OcrBody, @@ -467,6 +468,17 @@ class ProxyClient: ) ).info + def memory_summary_everywhere(self) -> Mapping[str, Result[MemorySummaryResponse]]: + return { + url: transport.get( + "/debug/memory/summary", + headers=transport.master, + params=NoBody(), + response_type=MemorySummaryResponse, + ) + for url, transport in self.replicas.items() + } + def read_back_everywhere[R: BaseModel]( self, path: str, diff --git a/tests/e2e/router/reliability_support.py b/tests/e2e/router/reliability_support.py index 1efcb1a045b..df2ff03aa4a 100644 --- a/tests/e2e/router/reliability_support.py +++ b/tests/e2e/router/reliability_support.py @@ -11,6 +11,8 @@ body, so a single long-lived proxy serves every reliability behavior. from __future__ import annotations +from collections.abc import Sequence + from pydantic import ValidationError from proxy_client import ProxyClient @@ -49,6 +51,13 @@ def create_bad_base_deployment(proxy: ProxyClient, name: str) -> str: ) +def create_never_benched_refusing_deployment(proxy: ProxyClient, name: str) -> str: + return proxy.create_model( + name, + LiteLLMParamsBody(model=REAL_MODEL, api_key=REAL_KEY, api_base="http://127.0.0.1:9/v1", cooldown_time=0), + ) + + def create_timeout_deployment(proxy: ProxyClient, name: str) -> str: """Register a deployment with a 1ms deadline the real backend always exceeds.""" return proxy.create_model(name, LiteLLMParamsBody(model=REAL_MODEL, api_key=REAL_KEY, timeout=0.001)) @@ -109,6 +118,7 @@ def chat_override( override: RouterSettingsOverride | None = None, stream: bool = False, cache: dict[str, bool] | None = {"no-cache": True}, + history: Sequence[ChatMessage] = (), ) -> StreamingResponse: """POST /chat/completions with an optional per-request router_settings_override, returning the raw outcome so tests read status, body, and reliability headers.""" @@ -117,7 +127,7 @@ def chat_override( headers=proxy.transport.bearer(key), json=ReliabilityChatBody( model=model, - messages=[ChatMessage(role="user", content=content)], + messages=[*history, ChatMessage(role="user", content=content)], max_tokens=512, stream=stream, router_settings_override=override, diff --git a/tests/e2e/router/test_reliability_memory_e2e.py b/tests/e2e/router/test_reliability_memory_e2e.py new file mode 100644 index 00000000000..17e3e1a1996 --- /dev/null +++ b/tests/e2e/router/test_reliability_memory_e2e.py @@ -0,0 +1,263 @@ +"""Live e2e: a few hundred requests that fail before any provider answers must not +grow the proxy's resident memory past a fixed budget once the proxy is warm. + +The regression this guards shipped in v1.100.0: every retry breadcrumb copied the +whole request and the copies nested into one router-global list, so a proxy under +retry-heavy failing traffic grew until it was OOM-killed. The traffic here has that +shape: a model group whose deployments refuse at the socket (an unreachable base +URL) with cooldown_time 0 so the router keeps retrying them, per-request retries, +and a fallback group that refuses the same way, each request carrying a long chat +transcript so every whole-request copy costs hundreds of containers instead of a +handful. Under the stack's cooldown policy a +deployment that fails a handful of times in a row is benched (a bad-credential 401 +included), the router answers "No deployments available" without retrying, and the +retry loop that leaks stops running; cooldown_time 0 keeps it running. + +Two identical phases run back to back. The first is the warmup that grows the +proxy's caches and allocator arenas to their steady state, the second is the one +the budget applies to, so a healthy proxy shows the second phase adding roughly +nothing while a leaking one adds a fixed amount per request. RSS is read through +/debug/memory/summary on every configured replica; a burst of failing calls leaves +a transient bulge of garbage that gc reclaims within seconds, so each checkpoint +samples until no new worker has answered for a settle window and keeps the lowest +reading per worker. The growth is judged per worker (by replica address, hostname +and pid, since pods in their own pid namespaces report the same pids) so each +worker is compared with itself, and the two checkpoints must see the same workers: +a single load-balanced address reaches the workers behind it one answer at a time, +and a worker that answered only one checkpoint would otherwise drop out of the +comparison, which is where a leaking worker could hide. + +RSS alone is a coarse gauge: on the release stack (spend logs storing prompts, +json logs, prometheus and otel callbacks) the same v1.100.0 breadcrumbs grew RSS +by only about 15 MB per 300 failing requests, while every failing request's stored +request snapshot carried a copy of the request per failed attempt, over 100 KB on +the first call and a couple of MB once the copies nested, against tens of KB with +the fix. So the first check sends one failing request before the phases, reads its +spend log back through /spend/logs, and holds the stored request body to a fixed +size budget: the deterministic catch for a breadcrumb that copies the whole +request. It runs before the phases because the leaking writer drops its own rows +under the phases' traffic (a queue budget hit, a recursion limit on the nested +copies), which would turn the size check into a missing-row check. +""" + +from __future__ import annotations + +import json +import time +from collections.abc import Mapping, Sequence +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass +from types import MappingProxyType +from typing import Final + +import pytest + +from complexity_router_client import ComplexityRouterClient +from e2e_config import ( + MEMORY_CONCURRENCY, + MEMORY_REQUESTS_PER_PHASE, + MEMORY_RETRIES_PER_REQUEST, + MEMORY_RSS_BUDGET_MB, + MEMORY_RSS_SAMPLE_INTERVAL_SECONDS, + MEMORY_RSS_SETTLE_SAMPLES, + MEMORY_STORED_REQUEST_BUDGET_KB, + MEMORY_TRANSCRIPT_TURNS, + unique_marker, +) +from e2e_http import unwrap +from lifecycle import ResourceManager +from models import ChatMessage, RouterSettingsOverride, SpendLogRow +from proxy_client import ProxyClient +from reliability_support import chat_override, create_never_benched_refusing_deployment + +pytestmark = pytest.mark.e2e + +DEPLOYMENTS_PER_GROUP: Final = 2 +RSS_SAMPLE_CAP: Final = 4 * MEMORY_RSS_SETTLE_SAMPLES + + +@dataclass(frozen=True, slots=True) +class FailedCall: + status_code: int + seconds: float + body_head: str + call_id: str | None + + +WorkerKey = tuple[str, str | None, int] + + +@dataclass(frozen=True, slots=True) +class RssReading: + replica: str + hostname: str | None + worker_pid: int + ram_usage_mb: float + + @property + def worker(self) -> WorkerKey: + return (self.replica, self.hostname, self.worker_pid) + + +@dataclass(frozen=True, slots=True) +class WorkerGrowth: + warm: RssReading + after: RssReading + + @property + def growth_mb(self) -> float: + return self.after.ram_usage_mb - self.warm.ram_usage_mb + + +def _register_refusing_group(proxy: ProxyClient, resources: ResourceManager, name: str) -> None: + for model_id in tuple(create_never_benched_refusing_deployment(proxy, name) for _ in range(DEPLOYMENTS_PER_GROUP)): + resources.defer(lambda model_id=model_id: proxy.delete_model(model_id)) + + +def _transcript(turns: int) -> tuple[ChatMessage, ...]: + return tuple( + ChatMessage(role=role, content=f"turn {turn} {role}") + for turn in range(turns) + for role in ("user", "assistant") + ) + + +TRANSCRIPT: Final = _transcript(MEMORY_TRANSCRIPT_TURNS) + + +def _fail_once(proxy: ProxyClient, key: str, model: str, override: RouterSettingsOverride) -> FailedCall: + started: Final = time.perf_counter() + resp: Final = chat_override( + proxy, key, model, f"memory regression {unique_marker()}", override=override, history=TRANSCRIPT + ) + return FailedCall(resp.status_code, time.perf_counter() - started, resp.body[:300], resp.call_id) + + +def _fail_many(proxy: ProxyClient, key: str, model: str, override: RouterSettingsOverride) -> tuple[FailedCall, ...]: + with ThreadPoolExecutor(max_workers=MEMORY_CONCURRENCY) as pool: + futures: Final = tuple( + pool.submit(_fail_once, proxy, key, model, override) for _ in range(MEMORY_REQUESTS_PER_PHASE) + ) + return tuple(future.result() for future in futures) + + +def _read_rss_everywhere_after_pause(proxy: ProxyClient) -> tuple[RssReading, ...]: + time.sleep(MEMORY_RSS_SAMPLE_INTERVAL_SECONDS) + return tuple( + RssReading(replica, body.hostname, body.worker_pid, body.memory.ram_usage_mb) + for replica, result in proxy.memory_summary_everywhere().items() + for body in (unwrap(result),) + if body.memory.ram_usage_mb is not None + ) + + +def _readings_until_no_new_worker( + proxy: ProxyClient, readings: tuple[RssReading, ...], samples: int, samples_since_new_worker: int +) -> tuple[RssReading, ...]: + if samples >= RSS_SAMPLE_CAP or samples_since_new_worker >= MEMORY_RSS_SETTLE_SAMPLES: + return readings + sample: Final = _read_rss_everywhere_after_pause(proxy) + known: Final = frozenset(reading.worker for reading in readings) + new_worker_answered: Final = any(reading.worker not in known for reading in sample) + return _readings_until_no_new_worker( + proxy, readings + sample, samples + 1, 0 if new_worker_answered else samples_since_new_worker + 1 + ) + + +def _settled_rss_per_worker(proxy: ProxyClient) -> Mapping[WorkerKey, RssReading]: + readings: Final = _readings_until_no_new_worker(proxy, (), 0, 0) + assert readings, "no /debug/memory/summary read carried ram_usage_mb, so the proxy cannot report its RSS" + return MappingProxyType( + { + worker: min((reading for reading in readings if reading.worker == worker), key=lambda r: r.ram_usage_mb) + for worker in {reading.worker for reading in readings} + } + ) + + +def _heaviest_worker_growth( + warm: Mapping[WorkerKey, RssReading], after: Mapping[WorkerKey, RssReading] +) -> WorkerGrowth: + assert warm.keys() == after.keys(), ( + f"the workers answering /debug/memory/summary changed between the checkpoints, so not every worker can " + f"be compared with itself: gone after the measured batch {sorted(warm.keys() - after.keys())} (a worker " + f"that died or was restarted under failing traffic, which is what an OOM kill looks like), first seen " + f"after it {sorted(after.keys() - warm.keys())} (the warm window never reached them, so they have no " + f"baseline; raise E2E_MEMORY_RSS_SETTLE_SAMPLES if the stack has more workers than the window covers)" + ) + return max((WorkerGrowth(warm[worker], after[worker]) for worker in warm), key=lambda growth: growth.growth_mb) + + +def _assert_every_call_failed_through_fallback(calls: Sequence[FailedCall], fallback: str) -> None: + served: Final = tuple(call for call in calls if call.status_code == 200) + assert not served, ( + f"{len(served)} of {len(calls)} calls came back 200, so they reached a provider and never " + f"exercised the retry loop: {served[0].body_head}" + ) + without_fallback: Final = tuple(call for call in calls if fallback not in call.body_head) + assert not without_fallback, ( + f"{len(without_fallback)} of {len(calls)} failures never named the fallback group {fallback}, " + f"so the request did not run through retries into the fallback: {without_fallback[0].body_head}" + ) + + +def _stored_request_kb(proxy: ProxyClient, call: FailedCall) -> float: + assert call.call_id, ( + f"the failing call carried no x-litellm-call-id header, so its spend log cannot be read back: {call.body_head}" + ) + rows: Final[Sequence[SpendLogRow]] = proxy.poll_logs_for_request_id(call.call_id) + assert rows, ( + f"no spend log row appeared for failing call {call.call_id} within the poll window: either the stack " + "writes no spend logs or its writer dropped the row, which the v1.100.0 one did once the stored " + "request outgrew the writer's queue budget" + ) + snapshot: Final = rows[0].proxy_server_request + assert snapshot, ( + f"spend log {call.call_id} stored no request body, so the stack is not running with " + "general_settings.store_prompts_in_spend_logs and the stored-request check would pass vacuously" + ) + return len(json.dumps(snapshot).encode()) / 1024 + + +class TestReliabilityMemory: + @pytest.mark.covers("reliability.perf.memory.under_slo") + def test_failing_requests_do_not_grow_rss_or_stored_request( + self, client: ComplexityRouterClient, resources: ResourceManager, scoped_key: str + ) -> None: + marker: Final = unique_marker() + primary: Final = f"reliability-memory-{marker}" + fallback: Final = f"reliability-memory-fb-{marker}" + _register_refusing_group(client.proxy, resources, primary) + _register_refusing_group(client.proxy, resources, fallback) + override: Final = RouterSettingsOverride( + num_retries=MEMORY_RETRIES_PER_REQUEST, fallbacks=[{primary: [fallback]}] + ) + + probe: Final = _fail_once(client.proxy, scoped_key, primary, override) + _assert_every_call_failed_through_fallback((probe,), fallback) + stored_kb: Final = _stored_request_kb(client.proxy, probe) + assert stored_kb <= MEMORY_STORED_REQUEST_BUDGET_KB, ( + f"the spend log of one failing request stored a {stored_kb:.0f} KB request body, past the " + f"{MEMORY_STORED_REQUEST_BUDGET_KB:.0f} KB budget for a {len(TRANSCRIPT)}-message transcript with " + f"{MEMORY_RETRIES_PER_REQUEST} retries and a fallback; the retry breadcrumbs are copying the whole " + f"request into the stored snapshot the way the v1.100.0 ones did" + ) + + warmup: Final = _fail_many(client.proxy, scoped_key, primary, override) + _assert_every_call_failed_through_fallback(warmup, fallback) + warm: Final = _settled_rss_per_worker(client.proxy) + + measured: Final = _fail_many(client.proxy, scoped_key, primary, override) + _assert_every_call_failed_through_fallback(measured, fallback) + after: Final = _settled_rss_per_worker(client.proxy) + + heaviest: Final = _heaviest_worker_growth(warm, after) + assert heaviest.growth_mb <= MEMORY_RSS_BUDGET_MB, ( + f"proxy RSS grew {heaviest.growth_mb:.1f} MB over a second batch of {MEMORY_REQUESTS_PER_PHASE} failing " + f"requests ({MEMORY_RETRIES_PER_REQUEST} retries each plus a fallback) after an identical warmup batch, " + f"past the {MEMORY_RSS_BUDGET_MB:.0f} MB budget: worker pid {heaviest.warm.worker_pid} on " + f"{heaviest.warm.hostname or 'an unnamed host'} behind {heaviest.warm.replica} settled at " + f"{heaviest.warm.ram_usage_mb:.1f} MB warm and " + f"{heaviest.after.ram_usage_mb:.1f} MB after; failing requests are leaking memory the way the " + f"v1.100.0 retry breadcrumbs did" + ) diff --git a/tests/test_litellm/proxy/common_utils/test_debug_utils.py b/tests/test_litellm/proxy/common_utils/test_debug_utils.py new file mode 100644 index 00000000000..163ea530be9 --- /dev/null +++ b/tests/test_litellm/proxy/common_utils/test_debug_utils.py @@ -0,0 +1,69 @@ +import os +import socket +from pathlib import Path + +import pytest + +from litellm.proxy._types import UserAPIKeyAuth +from litellm.proxy.common_utils.debug_utils import ( + PSUTIL_MISSING_ERROR, + _ProcFilesystemProcess, + _summary_process_memory, + get_memory_summary, +) + +PAGE_SIZE = 4096 +STATM_SIZE_PAGES = 100_000 +STATM_RESIDENT_PAGES = 30_000 +MEMINFO_TOTAL_KB = 1_000_000 + + +@pytest.fixture +def proc_process(tmp_path: Path) -> _ProcFilesystemProcess: + statm = tmp_path / "statm" + statm.write_text(f"{STATM_SIZE_PAGES} {STATM_RESIDENT_PAGES} 5000 1 0 20000 0\n") + meminfo = tmp_path / "meminfo" + meminfo.write_text( + f"MemTotal: {MEMINFO_TOTAL_KB} kB\nMemFree: 400000 kB\nMemAvailable: 600000 kB\n" + ) + return _ProcFilesystemProcess(statm_path=str(statm), meminfo_path=str(meminfo), page_size=PAGE_SIZE) + + +def test_proc_filesystem_process_reads_resident_and_virtual_bytes_from_statm( + proc_process: _ProcFilesystemProcess, +) -> None: + memory_info = proc_process.memory_info() + + assert memory_info.rss == STATM_RESIDENT_PAGES * PAGE_SIZE + assert memory_info.vms == STATM_SIZE_PAGES * PAGE_SIZE + + +def test_proc_filesystem_process_reports_share_of_meminfo_total(proc_process: _ProcFilesystemProcess) -> None: + expected_percent = STATM_RESIDENT_PAGES * PAGE_SIZE / (MEMINFO_TOTAL_KB * 1024) * 100 + + assert proc_process.memory_percent() == pytest.approx(expected_percent) + + +def test_summary_reports_rss_from_the_proc_filesystem(proc_process: _ProcFilesystemProcess) -> None: + memory, health_status = _summary_process_memory(proc_process) + + assert memory["ram_usage_mb"] == round(STATM_RESIDENT_PAGES * PAGE_SIZE / (1024 * 1024), 2) + assert memory["system_memory_percent"] == pytest.approx(12.0) + assert health_status == "healthy" + assert "error" not in memory + + +def test_summary_without_any_memory_source_names_psutil_and_reports_no_rss() -> None: + memory, health_status = _summary_process_memory(None) + + assert memory == {"error": PSUTIL_MISSING_ERROR} + assert health_status == "healthy" + + +@pytest.mark.asyncio +async def test_memory_summary_names_the_host_and_worker_that_answered() -> None: + summary = await get_memory_summary(UserAPIKeyAuth()) + + assert summary["hostname"] == socket.gethostname() + assert summary["worker_pid"] == os.getpid() + assert summary["memory"]["ram_usage_mb"] > 0 diff --git a/tests/test_litellm/proxy/test_component_allowlists.py b/tests/test_litellm/proxy/test_component_allowlists.py index 0fdb43d60da..3073908fa54 100644 --- a/tests/test_litellm/proxy/test_component_allowlists.py +++ b/tests/test_litellm/proxy/test_component_allowlists.py @@ -196,6 +196,20 @@ def test_gateway_drops_ui_and_swagger_mounts(): f"Mount {path} must not be served by the gateway" +def test_gateway_keeps_memory_summary_and_trims_the_other_debug_routes(): + """The gateway serves /debug/memory/summary, since the RSS that matters is the + serving worker's and the memory regression e2e test reads it on every gateway + replica; the heavier and mutating /debug/memory routes stay on the backend.""" + debug_memory_routes = { + getattr(r, "path"): r for r in app.router.routes if str(getattr(r, "path", "")).startswith("/debug/memory/") + } + assert {"/debug/memory/summary", "/debug/memory/details", "/debug/memory/gc/configure"} <= set(debug_memory_routes) + assert _is_gateway_route(debug_memory_routes["/debug/memory/summary"]), \ + "/debug/memory/summary must survive the gateway route trim" + for path in ("/debug/memory/details", "/debug/memory/gc/configure"): + assert not _is_gateway_route(debug_memory_routes[path]), f"{path} must not be served by the gateway" + + def test_every_app_mount_is_assigned_to_a_component(): """Every Mount on the proxy app must be consciously assigned to a component. diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index fae668ad0f1..7ed8df6815d 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -4169,6 +4169,7 @@ export interface paths { * * Returns: * - worker_pid: Process ID + * - hostname: Host (the pod on Kubernetes) the worker runs on * - status: Overall health based on memory usage * - memory: Process memory usage and RAM info * - caches: Cache item counts and descriptions