diff --git a/tests/integration/_support/process.py b/tests/integration/_support/process.py index 12780540380..23297b28584 100644 --- a/tests/integration/_support/process.py +++ b/tests/integration/_support/process.py @@ -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 diff --git a/tests/integration/configuration/test_multi_worker_slow_import_boot.py b/tests/integration/configuration/test_multi_worker_slow_import_boot.py new file mode 100644 index 00000000000..a90160e5e80 --- /dev/null +++ b/tests/integration/configuration/test_multi_worker_slow_import_boot.py @@ -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 diff --git a/tests/integration/routing/test_priority_scheduler_queue_cleanup.py b/tests/integration/routing/test_priority_scheduler_queue_cleanup.py index a4606554d34..3c38acadaaf 100644 --- a/tests/integration/routing/test_priority_scheduler_queue_cleanup.py +++ b/tests/integration/routing/test_priority_scheduler_queue_cleanup.py @@ -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,