mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-24 00:52:24 +00:00
f7ae9efad2
* test(e2e): hold every worker under an idle RSS budget before any traffic The harness reads /debug/memory/summary on every replica once at collection time, right after the readiness gate and before this pytest process sends any traffic, and the memory suite's first test fails when any worker idles past E2E_MEMORY_IDLE_RSS_BUDGET_MB (768 MB by default) or gives no reading at all. A v1.100.x worker with a database idled at 836-886 MB where v1.101.0rc1 idled at 544 MB on the same database: prisma-client-py's default recursive type depth generated 91k TypedDict classes that v1.101.0's recursive_type_depth = -1 cut to 19k. The budget starts at the rc1 reading plus headroom. * test(e2e): read idle RSS only when the idle budget test is selected Gate the collection-time /debug/memory/summary read on a selected test using the idle_rss fixture and skip it under --collect-only, so sessions that never run the idle budget test pay no round trip. Drop the markerless unit test file the e2e guide bans and assert live that every configured replica was measured * test(e2e): take the idle RSS read after collection settles Read every replica's RSS from a tryfirst pytest_collection_finish hook so -k and -m deselection has already run, and only when a selected test still asks for the idle_rss fixture and the run is not --collect-only * test(e2e): record the heaviest idle RSS reading as junit properties * test(e2e): attach the idle RSS properties from the harness's setup hook --------- Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
78 lines
2.9 KiB
Python
78 lines
2.9 KiB
Python
"""Per-worker RSS readings of the proxy through /debug/memory/summary.
|
|
|
|
One read goes to every configured replica (PROXY_REPLICA_URLS) under the master
|
|
key and answers from whichever worker behind that address took the connection;
|
|
the release stack runs one worker per gateway replica, so a read per replica is
|
|
a read per worker. A reading keys its worker by replica address, hostname, and
|
|
pid, since pods in their own pid namespaces report the same pids. A replica that
|
|
gives no reading (unreachable, a non-2xx, or a summary without ram_usage_mb) is
|
|
kept as a failure reason rather than dropped, so a test can fail on it by name
|
|
instead of passing on the replicas that did answer.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass
|
|
from typing import Final
|
|
|
|
from e2e_http import Result, Success
|
|
from models import MemorySummaryResponse
|
|
from proxy_client import ProxyClient
|
|
|
|
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)
|
|
|
|
@property
|
|
def where(self) -> str:
|
|
return f"worker pid {self.worker_pid} on {self.hostname or 'an unnamed host'} behind {self.replica}"
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RssCapture:
|
|
readings: tuple[RssReading, ...]
|
|
failures: tuple[str, ...]
|
|
|
|
@property
|
|
def heaviest(self) -> RssReading | None:
|
|
return max(self.readings, key=lambda reading: reading.ram_usage_mb, default=None)
|
|
|
|
@property
|
|
def junit_properties(self) -> tuple[tuple[str, object], ...]:
|
|
heaviest: Final = self.heaviest
|
|
if heaviest is None:
|
|
return ()
|
|
return (("idle_rss_heaviest_mb", heaviest.ram_usage_mb), ("idle_rss_heaviest_worker", heaviest.where))
|
|
|
|
|
|
def _outcome(replica: str, result: Result[MemorySummaryResponse]) -> RssReading | str:
|
|
match result:
|
|
case Success(data=body) if body.memory.ram_usage_mb is not None:
|
|
return RssReading(replica, body.hostname, body.worker_pid, body.memory.ram_usage_mb)
|
|
case Success(data=body):
|
|
return f"{replica} answered /debug/memory/summary without ram_usage_mb: {body.memory.error}"
|
|
case _:
|
|
return f"{replica} gave no /debug/memory/summary reading: {result}"
|
|
|
|
|
|
def rss_capture(summaries: Mapping[str, Result[MemorySummaryResponse]]) -> RssCapture:
|
|
outcomes: Final = tuple(_outcome(replica, result) for replica, result in summaries.items())
|
|
return RssCapture(
|
|
readings=tuple(outcome for outcome in outcomes if isinstance(outcome, RssReading)),
|
|
failures=tuple(outcome for outcome in outcomes if isinstance(outcome, str)),
|
|
)
|
|
|
|
|
|
def read_rss_everywhere(proxy: ProxyClient, *, timeout: float | None = None) -> RssCapture:
|
|
return rss_capture(proxy.memory_summary_everywhere(timeout=timeout))
|