mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-09 03:18:44 +00:00
test(integration): boot owned proxies with a readiness-sized worker healthcheck budget (#45292)
* test(integration): boot owned proxies with a readiness-sized worker healthcheck budget * test(integration): keep owned_proxy_process's extra_arguments for cells that pass their own flags --------- Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot]
GitHub
parent
91f36f1774
commit
9df3d4c96b
3 changed files with 55 additions and 23 deletions
|
|
@ -260,6 +260,8 @@ def owned_proxy_process(
|
|||
"127.0.0.1",
|
||||
"--num_workers",
|
||||
str(workers),
|
||||
"--timeout_worker_healthcheck",
|
||||
str(int(graceful_stop_seconds())),
|
||||
*database_setup,
|
||||
*extra_arguments,
|
||||
)
|
||||
|
|
@ -294,6 +296,8 @@ def owned_gateway_image(
|
|||
str(workers),
|
||||
"--host",
|
||||
"127.0.0.1",
|
||||
"--timeout-worker-healthcheck",
|
||||
str(int(graceful_stop_seconds())),
|
||||
)
|
||||
launch: Final = _launch_until_bound(command, root, environment, output, _PORT_ATTEMPTS)
|
||||
try:
|
||||
|
|
@ -397,9 +401,7 @@ class UpstreamSlot:
|
|||
|
||||
__slots__ = ("certificate", "directory", "port", "process", "root")
|
||||
|
||||
def __init__(
|
||||
self, directory: Path, port: int, root: Path, certificate: UpstreamCertificate | None = None
|
||||
) -> None:
|
||||
def __init__(self, directory: Path, port: int, root: Path, certificate: UpstreamCertificate | None = None) -> None:
|
||||
self.directory = directory
|
||||
self.port = port
|
||||
self.root = root
|
||||
|
|
|
|||
|
|
@ -0,0 +1,45 @@
|
|||
import os
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
from integration._support.client import Gateway, eventually
|
||||
from integration._support.process import graceful_stop_seconds, owned_proxy_process
|
||||
|
||||
WORKERS: Final = 2
|
||||
DEPLOYMENT_HEALTHCHECK_SECONDS: Final = 5
|
||||
WORKER_IMPORT_DELAY_SECONDS: Final = 3 * DEPLOYMENT_HEALTHCHECK_SECONDS
|
||||
STARTED_WORKER: Final = re.compile(r"Started server process \[(\d+)\]")
|
||||
DIED_WORKER: Final = re.compile(r"Child process \[(\d+)\] died")
|
||||
SLOW_WORKER_HOOK: Final = f"""\
|
||||
import sys
|
||||
import time
|
||||
|
||||
if "--multiprocessing-fork" in sys.argv:
|
||||
time.sleep({WORKER_IMPORT_DELAY_SECONDS})
|
||||
"""
|
||||
|
||||
|
||||
def _slow_worker_environment(directory: Path) -> dict[str, str]:
|
||||
hook: Final = directory / "slow_worker"
|
||||
hook.mkdir()
|
||||
(hook / "sitecustomize.py").write_text(SLOW_WORKER_HOOK)
|
||||
return {
|
||||
"PYTHONPATH": os.pathsep.join((str(hook), os.environ.get("PYTHONPATH", ""))),
|
||||
"TIMEOUT_WORKER_HEALTHCHECK": str(DEPLOYMENT_HEALTHCHECK_SECONDS),
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.timeout(2 * graceful_stop_seconds() + 60)
|
||||
def test_owned_proxy_workers_outlast_the_deployments_healthcheck_default(gateway: Gateway, tmp_path: Path) -> None:
|
||||
with owned_proxy_process(gateway, tmp_path, _slow_worker_environment(tmp_path), workers=WORKERS) as owned:
|
||||
started: Final = eventually(
|
||||
lambda: STARTED_WORKER.findall(owned.log.read_text()),
|
||||
lambda pids: len(pids) >= WORKERS,
|
||||
seconds=graceful_stop_seconds(),
|
||||
)
|
||||
assert owned.gateway.request("GET", "/health/readiness").status_code == 200
|
||||
log: Final = owned.log.read_text()
|
||||
assert len(started) == WORKERS, log
|
||||
assert DIED_WORKER.findall(log) == [], log
|
||||
|
|
@ -46,7 +46,6 @@ USAGE: Final[dict[str, JsonValue]] = {"prompt_tokens": 5, "completion_tokens": 3
|
|||
STARTED_WORKER: Final = re.compile(r"Started server process \[(\d+)\]")
|
||||
OWNED_CELL_TIMEOUT: Final = 2 * graceful_stop_seconds() + 120
|
||||
PAIR_CELL_TIMEOUT: Final = 3 * graceful_stop_seconds() + 120
|
||||
WORKER_HEALTHCHECK_ARGUMENTS: Final = ("--timeout_worker_healthcheck", str(int(graceful_stop_seconds())))
|
||||
PINNED_CONNECTION_TIMEOUT_SECONDS: Final = 30
|
||||
PINNED_LIMITS: Final = httpx.Limits(max_connections=1, max_keepalive_connections=1, keepalive_expiry=30)
|
||||
ENDPOINTS: Final[tuple[Endpoint, ...]] = (
|
||||
|
|
@ -541,11 +540,7 @@ def test_in_memory_queue_forgets_served_requests_before_a_cooldown(gateway: Gate
|
|||
with ExitStack() as stack:
|
||||
wire: Final = stack.enter_context(wire_server(answering_model_discovery(upstream.respond)))
|
||||
config: Final = owned_config(tmp_path, wire, (INMEM_GROUP,), cooldown_settings(None))
|
||||
owned: Final = stack.enter_context(
|
||||
owned_proxy_process(
|
||||
gateway, tmp_path, {}, config=config, workers=2, extra_arguments=WORKER_HEALTHCHECK_ARGUMENTS
|
||||
)
|
||||
)
|
||||
owned: Final = stack.enter_context(owned_proxy_process(gateway, tmp_path, {}, config=config, workers=2))
|
||||
worker: Final = pinned(owned.gateway, stack)
|
||||
served: Final = new_marker()
|
||||
assert_served(post(worker, "/v1/chat/completions", chat_body(INMEM_GROUP, served, priority=1)), served)
|
||||
|
|
@ -580,15 +575,9 @@ def pair(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Pair]:
|
|||
wire: Final = stack.enter_context(wire_server(answering_model_discovery(upstream.respond)))
|
||||
config: Final = owned_config(directory, wire, PAIR_GROUPS, cooldown_settings(cache), cancel_on_disconnect=True)
|
||||
overrides: Final = {"REDIS_HOST": cache.host, "REDIS_PORT": str(cache.port)}
|
||||
first: Final = stack.enter_context(
|
||||
owned_proxy_process(
|
||||
gateway, directory, overrides, config=config, workers=2, extra_arguments=WORKER_HEALTHCHECK_ARGUMENTS
|
||||
)
|
||||
)
|
||||
first: Final = stack.enter_context(owned_proxy_process(gateway, directory, overrides, config=config, workers=2))
|
||||
second: Final = stack.enter_context(
|
||||
owned_proxy_process(
|
||||
gateway, directory, overrides, config=config, workers=2, extra_arguments=WORKER_HEALTHCHECK_ARGUMENTS
|
||||
)
|
||||
owned_proxy_process(gateway, directory, overrides, config=config, workers=2)
|
||||
)
|
||||
yield Pair(first.gateway, second.gateway, cache, wire, upstream)
|
||||
|
||||
|
|
@ -797,9 +786,7 @@ def test_prioritized_requests_survive_a_redis_outage(gateway: Gateway, tmp_path:
|
|||
"REDIS_PORT": str(cache.port),
|
||||
"REDIS_CIRCUIT_BREAKER_RECOVERY_TIMEOUT": "1",
|
||||
}
|
||||
with owned_proxy_process(
|
||||
gateway, tmp_path, overrides, config=config, workers=2, extra_arguments=WORKER_HEALTHCHECK_ARGUMENTS
|
||||
) as owned:
|
||||
with owned_proxy_process(gateway, tmp_path, overrides, config=config, workers=2) as owned:
|
||||
before: Final = assert_all_served(burst(owned.gateway, OUTAGE_GROUP, 12))
|
||||
cache.stop()
|
||||
during: Final = assert_all_served(burst(owned.gateway, OUTAGE_GROUP, 12))
|
||||
|
|
@ -826,9 +813,7 @@ def test_sibling_worker_keeps_serving_prioritized_requests_after_a_worker_is_kil
|
|||
with owned_redis(tmp_path) as cache, wire_server(answering_model_discovery(upstream.respond)) as wire:
|
||||
config: Final = owned_config(tmp_path, wire, (KILL_GROUP,), redis_settings(cache))
|
||||
overrides: Final = {"REDIS_HOST": cache.host, "REDIS_PORT": str(cache.port)}
|
||||
with owned_proxy_process(
|
||||
gateway, tmp_path, overrides, config=config, workers=2, extra_arguments=WORKER_HEALTHCHECK_ARGUMENTS
|
||||
) as owned:
|
||||
with owned_proxy_process(gateway, tmp_path, overrides, config=config, workers=2) as owned:
|
||||
workers: Final = eventually(
|
||||
lambda: tuple(int(found.group(1)) for found in STARTED_WORKER.finditer(owned.log.read_text())),
|
||||
lambda pids: len(pids) == 2,
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue