Skip to content

GitLab

"lib/llm/vscode:/vscode.git/clone" did not exist on "95a750f48b4b8a81803aa141364a6de5a1920ed4"
Unverified Commit 59535682 authored by Jacky's avatar Jacky Committed by GitHub
Browse files

test: Add request migration graceful shutdown E2E test (#3585) 


Signed-off-by: default avatarJacky <18255193+kthui@users.noreply.github.com>
parent 8c8922b0
Show whitespace changes
Inline Side-by-side
Showing with 630 additions and 109 deletions
tests/fault_tolerance/README.md
View file @ 59535682
# Fault Tolerance Tests
This directory contains end-to-end tests for Dynamo's fault tolerance capabilities.
## Tests
### `test_request_migration.py`
Tests worker fault tolerance with migration support using the `test_request_migration_vllm` function. This test:
0. Downloads the DeepSeek-R1-Distill-Llama-8B model from HuggingFace if not already cached
1. Starts a Dynamo frontend using `python -m dynamo.frontend` with round-robin routing
2. Starts 2 workers sequentially using `python3 -m dynamo.vllm` with specific configuration:
- Model: `deepseek-ai/DeepSeek-R1-Distill-Llama-8B`
- `--enforce-eager`, `--gpu-memory-utilization 0.45`
- `--max-model-len 8192`, `--migration-limit 3`
3. Waits for both workers to be fully ready (health check returns "ready" status)
4. Sends a test request ("Who are you?", 100 tokens) to determine which worker handles requests
5. Determines primary/backup worker roles based on round-robin routing and log analysis
6. Sends a long completion request ("Tell me a long long long story about yourself?", 8000 tokens) in a separate thread
7. Waits 0.5 seconds, then kills the primary worker using SIGKILL process group termination
8. Verifies the request completes successfully despite the worker failure (with 240s timeout)
9. Checks that the frontend logs contain "Stream disconnected... recreating stream..." indicating migration occurred
### `test_request_cancellation.py`
Tests request cancellation functionality across multiple API endpoints and deployment configurations. Contains three test functions:
#### `test_request_cancellation_vllm`
Tests basic request cancellation with a single worker:
0. Downloads the DeepSeek-R1-Distill-Llama-8B model from HuggingFace if not already cached
1. Starts a Dynamo frontend using `python -m dynamo.frontend` with debug logging enabled
2. Starts a single worker using `python3 -m dynamo.vllm` with specific configuration:
- Model: `deepseek-ai/DeepSeek-R1-Distill-Llama-8B`
- `--enforce-eager`, `--gpu-memory-utilization 0.45`, `--max-model-len 8192`, `--migration-limit 3`
- Debug logging enabled on port 8081
3. Tests request cancellation across three scenarios:
- **Completion API**: `/v1/completions` endpoint cancellation
- **Chat Completion API (non-streaming)**: `/v1/chat/completions` endpoint cancellation
- **Chat Completion API (streaming)**: `/v1/chat/completions` with streaming cancellation
4. For each scenario:
- Sends a long request with 1-second timeout to trigger cancellation
- Validates that cancellation messages appear in both frontend and worker logs
- Uses incremental log offset tracking to avoid false positives from previous tests
5. Checks for specific cancellation patterns:
- Frontend log: "issued control message Kill to sender"
- Worker log: "Aborted Request ID: <request_id>" matching the "New Request ID: <request_id>"
#### `test_request_cancellation_vllm_decode`
Tests request cancellation during disaggregated decode phase:
0. Downloads the DeepSeek-R1-Distill-Llama-8B model from HuggingFace if not already cached
1. Starts a Dynamo frontend using `python -m dynamo.frontend` with debug logging enabled
2. Starts a prefill worker using `python3 -m dynamo.vllm --is-prefill-worker` on port 8082
3. Starts a decode worker using `python3 -m dynamo.vllm` on port 8081
4. Tests completion request cancellation in the disaggregated setup
5. Validates cancellation messages appear in prefill worker, decode worker, and frontend logs
6. Checks for specific patterns:
- Frontend log: "issued control message Kill to sender"
- Decode worker log: "Aborted Request ID: <request_id>"
- Prefill worker log: "New Prefill Request ID: <request_id>"
#### `test_request_cancellation_vllm_prefill`
Tests request cancellation during disaggregated prefill phase:
- (Skipped until request cancellation can cancel before receiving the first response)
## Prerequisites
- vLLM backend installed
- NATS and etcd services running (provided by `runtime_services` fixture)
- Access to DeepSeek-R1-Distill-Llama-8B model (automatically downloaded from HuggingFace)
- Sufficient GPU memory
## Running the Tests
To run the fault tolerance tests:
## Migration Tests
The migration directory contains tests for worker fault tolerance with migration support.
### Test Matrix
| Test | Shutdown Method | Migration Enabled | Expected Result | Verification |
|------|----------------|-------------------|-----------------|--------------|
| `test_request_migration_vllm_worker_failure` | SIGKILL (immediate) | Yes (default) | Request succeeds | "Stream disconnected... recreating stream..." in logs |
| `test_request_migration_vllm_graceful_shutdown` | SIGTERM (10s timeout) | Yes (default) | Request succeeds | "Stream disconnected... recreating stream..." in logs |
| `test_no_request_migration_vllm_worker_failure` | SIGKILL (immediate) | No (migration_limit=0) | Request fails (500) | "Migration limit exhausted" in logs |
| `test_no_request_migration_vllm_graceful_shutdown` | SIGTERM (10s timeout) | No (migration_limit=0) | Request fails (500) | "Migration limit exhausted" in logs |
### Common Test Flow
All migration tests follow this pattern:
1. Start a Dynamo frontend with round-robin routing
2. Start 2 vLLM workers sequentially
3. Send a long completion request (max_tokens=8192) in a separate daemon thread
4. Use parallel polling to determine which worker received the request (checks for "New Request ID:" in logs)
5. Terminate the worker processing the request (method varies by test)
6. Validate the request outcome (success or failure based on migration setting)
7. Verify migration behavior in frontend logs
**Run examples:**
```bash
# Run all fault tolerance tests
pytest -m "e2e and vllm" /workspace/tests/fault_tolerance
# With migration enabled
pytest tests/fault_tolerance/migration/test_vllm.py::test_request_migration_vllm_worker_failure -v -s
pytest tests/fault_tolerance/migration/test_vllm.py::test_request_migration_vllm_graceful_shutdown -v -s
# With migration disabled
pytest tests/fault_tolerance/migration/test_vllm.py::test_no_request_migration_vllm_worker_failure -v -s
pytest tests/fault_tolerance/migration/test_vllm.py::test_no_request_migration_vllm_graceful_shutdown -v -s
```
## Cancellation Tests
# Run specific test functions with debug logging
pytest /workspace/tests/fault_tolerance/test_request_migration.py::test_request_migration_vllm -v -s
pytest /workspace/tests/fault_tolerance/test_request_cancellation.py::test_request_cancellation_vllm -v -s
pytest /workspace/tests/fault_tolerance/test_request_cancellation.py::test_request_cancellation_vllm_decode -v -s
The cancellation directory contains tests for request cancellation functionality across multiple
API endpoints, backends, and deployment configurations.
### Test Overview by Backend
#### vLLM Cancellation Tests
| Test | Mode | Cancellation Phase | Request Type | Setup |
|------|------|-------------------|--------------|-------|
| `test_request_cancellation_vllm_aggregated` | Aggregated | During generation | 3 scenarios: completion, chat, streaming chat | 1 worker |
| `test_request_cancellation_vllm_decode_cancel` | Disaggregated | Remote decode | Streaming chat (5 responses read) | Prefill + Decode workers |
| `test_request_cancellation_vllm_remote_prefill_cancel` | Disaggregated | Remote prefill | Completion (long prompt) | Prefill + Decode workers |
**Run examples:**
```bash
pytest tests/fault_tolerance/cancellation/test_vllm.py::test_request_cancellation_vllm_aggregated -v -s
pytest tests/fault_tolerance/cancellation/test_vllm.py::test_request_cancellation_vllm_decode_cancel -v -s
pytest tests/fault_tolerance/cancellation/test_vllm.py::test_request_cancellation_vllm_remote_prefill_cancel -v -s
```
## Test Markers
#### TRT-LLM Cancellation Tests
- `@pytest.mark.e2e`: End-to-end test
- `@pytest.mark.vllm`: Requires vLLM backend
- `@pytest.mark.gpu_1`: Requires single GPU access
- `@pytest.mark.slow`: Known to be slow (due to model loading and inference)
| Test | Mode | Strategy | Cancellation Phase | Request Type | Setup |
|------|------|----------|-------------------|--------------|-------|
| `test_request_cancellation_trtllm_aggregated` | Aggregated | N/A | During generation | 3 scenarios: completion, chat, streaming chat | 1 worker (prefill_and_decode) |
| `test_request_cancellation_trtllm_decode_first_decode_cancel` | Disaggregated | Decode-first | Remote decode | Streaming chat (5 responses read) | Prefill + Decode workers |
| `test_request_cancellation_trtllm_decode_first_remote_prefill_cancel` | Disaggregated | Decode-first | Remote prefill | Completion (long prompt) | Prefill + Decode workers |
| `test_request_cancellation_trtllm_prefill_first_prefill_cancel` | Disaggregated | Prefill-first | Local prefill | Completion (long prompt) | Decode + Prefill workers |
| `test_request_cancellation_trtllm_prefill_first_remote_decode_cancel` | Disaggregated | Prefill-first | Remote decode | Streaming chat (5 responses read) | Decode + Prefill workers |
## Environment Variables
**Run examples:**
```bash
pytest tests/fault_tolerance/cancellation/test_trtllm.py::test_request_cancellation_trtllm_aggregated -v -s
pytest tests/fault_tolerance/cancellation/test_trtllm.py::test_request_cancellation_trtllm_decode_first_decode_cancel -v -s
# ... (other tests follow same pattern)
```
- `DYN_LOG`: Set to `debug` or `trace` for verbose logging (automatically set to `debug` by worker processes)
- `CUDA_VISIBLE_DEVICES`: Control which GPUs are used for testing
#### SGLang Cancellation Tests
## Expected Test Duration
| Test | Mode | Cancellation Phase | Request Type | Setup | Notes |
|------|------|-------------------|--------------|-------|-------|
| `test_request_cancellation_sglang_aggregated` | Aggregated | During generation | 3 scenarios: completion, chat, streaming chat (1 response read) | 1 worker | ⚠️ Flaky: SGLang prefill cancellation issues |
| `test_request_cancellation_sglang_decode_cancel` | Disaggregated | Remote decode | Streaming chat (1 response read) | Decode + Prefill workers | Requires 2 GPUs |
The tests typically take 2-3 minutes to complete each, including:
- Model download/loading time (if not cached) - can take 1-2 minutes for first run
- Worker startup and registration
- Request processing and response validation
- Worker failure simulation and migration (for migration test) / Request cancellation validation (for cancellation tests)
- Cleanup
**Run examples:**
```bash
pytest tests/fault_tolerance/cancellation/test_sglang.py::test_request_cancellation_sglang_aggregated -v -s
pytest tests/fault_tolerance/cancellation/test_sglang.py::test_request_cancellation_sglang_decode_cancel -v -s
```
## Troubleshooting
### Common Cancellation Test Pattern
If tests fail:
1. Start frontend and workers (configuration varies by test)
2. Send request (type varies by test scenario)
3. Poll for request ID in worker logs
4. For streaming: read N responses before cancellation
5. Cancel the request via API
6. Verify cancellation messages in worker and frontend logs
1. Check that NATS and etcd services are running
2. Verify vLLM backend is properly installed
3. Ensure sufficient GPU memory is available
4. Check internet connectivity for model download from HuggingFace
5. Review test logs for specific error messages
6. Verify that the DeepSeek-R1-Distill-Llama-8B model can be accessed
7. For cancellation tests: Check that timeout-based cancellation is working properly and cancellation patterns appear in logs
8. For migration tests: Verify worker process termination and stream recreation behavior
9. For disaggregated cancellation tests: Ensure both prefill and decode workers are properly started and cancellation works across the disaggregated setup
**Verification patterns:**
- Aggregated mode: "Aborted Request ID" in worker logs
- Remote prefill: "Aborted Request ID" in prefill, "Aborted Remote Request ID" in decode
- Remote decode: "Aborted Request ID" in decode, "Aborted Remote Request ID" in prefill
tests/fault_tolerance/cancellation/utils.py
View file @ 59535682
......@@ -350,7 +350,7 @@ def poll_for_pattern(
while iteration < max_iterations:
# Read the process log
log_content = read_log_content(process._log_path)
log_content = read_log_content(process.log_path)
new_content = log_content[current_offset:]
# Look for the pattern
......
tests/fault_tolerance/test_request_migration.py → tests/fault_tolerance/migration/test_vllm.py
View file @ 59535682
......@@ -3,50 +3,31 @@
import logging
import os
import queue
import shutil
import threading
import time
import pytest
import requests
from tests.utils.constants import FAULT_TOLERANCE_MODEL_NAME
from tests.utils.engine_process import FRONTEND_PORT
from tests.utils.managed_process import ManagedProcess, terminate_process_tree
from tests.utils.payloads import check_models_api
logger = logging.getLogger(__name__)
class DynamoFrontendProcess(ManagedProcess):
"""Process manager for Dynamo frontend"""
def __init__(self, request):
command = ["python", "-m", "dynamo.frontend", "--router-mode", "round-robin"]
# Import utilities from the refactored utils module
from .utils import (
DynamoFrontendProcess,
determine_request_receiving_worker,
start_completion_request,
validate_completion_response,
verify_migration_occurred,
)
log_dir = f"{request.node.name}_frontend"
# Clean up any existing log directory from previous runs
try:
shutil.rmtree(log_dir)
logger.info(f"Cleaned up existing log directory: {log_dir}")
except FileNotFoundError:
# Directory doesn't exist, which is fine
pass
super().__init__(
command=command,
display_output=True,
terminate_existing=True,
log_dir=log_dir,
)
logger = logging.getLogger(__name__)
class DynamoWorkerProcess(ManagedProcess):
"""Process manager for Dynamo worker with vLLM backend"""
def __init__(self, request, worker_id: str):
def __init__(self, request, worker_id: str, migration_limit: int = 3):
self.worker_id = worker_id
command = [
......@@ -61,7 +42,7 @@ class DynamoWorkerProcess(ManagedProcess):
"--max-model-len",
"8192",
"--migration-limit",
"3",
str(migration_limit),
]
# Set debug logging environment
......@@ -117,254 +98,232 @@ class DynamoWorkerProcess(ManagedProcess):
return False
def send_completion_request(
prompt: str, max_tokens: int, timeout: int = 120
) -> requests.Response:
"""Send a completion request to the frontend"""
payload = {
"model": FAULT_TOLERANCE_MODEL_NAME,
"prompt": prompt,
"max_tokens": max_tokens,
}
@pytest.mark.vllm
@pytest.mark.gpu_1
@pytest.mark.e2e
@pytest.mark.model(FAULT_TOLERANCE_MODEL_NAME)
def test_request_migration_vllm_worker_failure(
request, runtime_services, predownload_models, set_ucx_tls_no_mm
):
"""
End-to-end test for worker fault tolerance with migration support.
headers = {"Content-Type": "application/json"}
This test verifies that when a worker is killed during request processing,
the system can handle the failure gracefully and migrate the request to
another worker.
"""
logger.info(
f"Sending completion request with prompt: '{prompt[:50]}...' and max_tokens: {max_tokens}"
)
# Step 1: Start the frontend
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
try:
response = requests.post(
"http://localhost:8000/v1/completions",
headers=headers,
json=payload,
timeout=timeout,
)
logger.info(f"Received response with status code: {response.status_code}")
return response
except requests.exceptions.Timeout:
logger.error(f"Request timed out after {timeout} seconds")
raise
except requests.exceptions.RequestException as e:
logger.error(f"Request failed with error: {e}")
raise
def validate_openai_response(response: requests.Response) -> None:
"""Validate that the response is a proper OpenAI completion response"""
assert (
response.status_code == 200
), f"Request failed with status {response.status_code}: {response.text}"
# Step 2: Start 2 workers sequentially
try:
data = response.json()
except ValueError:
pytest.fail(f"Response is not valid JSON: {response.text}")
# Start worker1 first and wait for it to be ready
logger.info("Starting worker 1...")
with DynamoWorkerProcess(request, "worker1") as worker1:
logger.info(f"Worker 1 PID: {worker1.get_pid()}")
# Validate OpenAI completion response structure
assert "choices" in data, f"Response missing 'choices' field: {data}"
assert len(data["choices"]) > 0, f"Response has empty 'choices': {data}"
assert "text" in data["choices"][0], f"Response choice missing 'text' field: {data}"
assert data["choices"][0]["text"], f"Response text is empty: {data}"
with DynamoWorkerProcess(request, "worker2") as worker2:
logger.info(f"Worker 2 PID: {worker2.get_pid()}")
# Step 3: Send the request
request_thread, response_list = start_completion_request()
# Step 4: Use polling to determine which worker received the request
worker, worker_name = determine_request_receiving_worker(
worker1, worker2, receiving_pattern="Decode Request ID: "
)
# Step 5: Kill the worker that has the request
logger.info(
f"Received valid completion response: {data['choices'][0]['text'][:100]}..."
f"Killing {worker_name} with PID {worker.get_pid()} processing the request"
)
terminate_process_tree(worker.get_pid(), immediate_kill=True, timeout=0)
# Step 6: Validate the completion response
validate_completion_response(request_thread, response_list)
# Step 7: Verify migration occurred
verify_migration_occurred(frontend)
def check_worker_received_request(worker_process: DynamoWorkerProcess) -> bool:
"""Check if the worker logs contain request ID message indicating it received a request.
Supports multiple backend patterns:
- vLLM: "Decode Request ID:" or "Prefill Request ID:"
- SGLang/TensorRT-LLM: "New Request ID:"
@pytest.mark.vllm
@pytest.mark.gpu_1
@pytest.mark.e2e
@pytest.mark.model(FAULT_TOLERANCE_MODEL_NAME)
def test_request_migration_vllm_graceful_shutdown(
request, runtime_services, predownload_models, set_ucx_tls_no_mm
):
"""
log_path = worker_process._log_path
if log_path and os.path.exists(log_path):
try:
with open(log_path, "r") as f:
log_content = f.read()
# Check for any of the supported patterns
return (
"New Request ID: " in log_content
or "Decode Request ID: " in log_content
or "Prefill Request ID: " in log_content
)
except Exception as e:
logger.warning(f"Could not read worker log file {log_path}: {e}")
return False
End-to-end test for worker fault tolerance with graceful shutdown and migration support.
This test verifies that when a worker receives a graceful shutdown signal (SIGTERM)
during request processing, the system can handle the shutdown gracefully and migrate
the request to another worker. Unlike the abrupt kill test, this simulates a more
controlled shutdown scenario where the worker has time to clean up and notify the
system about its shutdown.
"""
def determine_worker_roles(worker1: DynamoWorkerProcess, worker2: DynamoWorkerProcess):
"""Determine primary and backup workers based on which worker handled the test request"""
worker1_received_test = check_worker_received_request(worker1)
worker2_received_test = check_worker_received_request(worker2)
if worker1_received_test and not worker2_received_test:
primary_worker = (worker2, "Worker 2")
backup_worker = (worker1, "Worker 1")
logger.info("Test request was handled by Worker 1")
return primary_worker, backup_worker
elif worker2_received_test and not worker1_received_test:
primary_worker = (worker1, "Worker 1")
backup_worker = (worker2, "Worker 2")
logger.info("Test request was handled by Worker 2")
return primary_worker, backup_worker
else:
pytest.fail(
f"Could not determine which worker handled the test request. Worker1: {worker1_received_test}, Worker2: {worker2_received_test}"
)
# Step 1: Start the frontend
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start 2 workers sequentially
with DynamoWorkerProcess(request, "worker1") as worker1:
logger.info(f"Worker 1 PID: {worker1.get_pid()}")
def start_completion_request(primary_worker_name: str):
"""
Start a request in a separate thread.
with DynamoWorkerProcess(request, "worker2") as worker2:
logger.info(f"Worker 2 PID: {worker2.get_pid()}")
Args:
primary_worker_name: Name of the primary worker expected to handle the request
# Step 3: Send the request
request_thread, response_list = start_completion_request()
Returns:
tuple: (request_thread, response_queue)
"""
response_queue: queue.Queue[requests.Response] = queue.Queue()
# Step 4: Use polling to determine which worker received the request
worker, worker_name = determine_request_receiving_worker(
worker1, worker2, receiving_pattern="Decode Request ID: "
)
def send_formal_request():
response = send_completion_request(
"Tell me a long long long story about yourself?",
8000,
timeout=240, # Extended timeout for long request
# Step 5: Gracefully shutdown the worker that has the request
logger.info(
f"Gracefully shutting down {worker_name} with PID {worker.get_pid()} processing the request"
)
terminate_process_tree(
worker.get_pid(), immediate_kill=False, timeout=10
)
response_queue.put(response)
request_thread = threading.Thread(target=send_formal_request)
request_thread.start()
# Step 6: Validate the completion response
validate_completion_response(request_thread, response_list)
return request_thread, response_queue
# Step 7: Verify migration occurred during graceful shutdown
verify_migration_occurred(frontend)
def validate_completion_response(
request_thread: threading.Thread, response_queue: queue.Queue
@pytest.mark.vllm
@pytest.mark.gpu_1
@pytest.mark.e2e
@pytest.mark.model(FAULT_TOLERANCE_MODEL_NAME)
def test_no_request_migration_vllm_worker_failure(
request, runtime_services, predownload_models, set_ucx_tls_no_mm
):
"""
Wait for and validate the completion response after worker failure.
End-to-end test for worker fault tolerance with migration disabled.
Args:
request_thread: The thread running the completion request
response_queue: Queue containing the response from the request
This test verifies that when migration is disabled (migration_limit=0) and a worker
is killed during request processing, the request fails as expected without migration.
This is the opposite behavior of test_request_migration_vllm_worker_failure.
"""
request_thread.join(timeout=300)
if request_thread.is_alive():
pytest.fail("Request did not complete within timeout")
# Get the response
if response_queue.empty():
pytest.fail("No response received for request")
response = response_queue.get()
# Step 1: Start the frontend
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Validate the response
validate_openai_response(response)
logger.info("✓ Request completed successfully after worker failure")
# Step 2: Start 2 workers sequentially with migration disabled
logger.info("Starting worker 1 with migration disabled...")
with DynamoWorkerProcess(request, "worker1", migration_limit=0) as worker1:
logger.info(f"Worker 1 PID: {worker1.get_pid()}")
with DynamoWorkerProcess(request, "worker2", migration_limit=0) as worker2:
logger.info(f"Worker 2 PID: {worker2.get_pid()}")
def verify_migration_occurred(frontend_process: DynamoFrontendProcess) -> None:
"""
Verify that migration occurred by checking frontend logs for stream disconnection message.
# Step 3: Send the request
request_thread, response_list = start_completion_request()
Args:
frontend_process: The frontend process to check logs for
# Step 4: Use polling to determine which worker received the request
worker, worker_name = determine_request_receiving_worker(
worker1, worker2, receiving_pattern="Decode Request ID: "
)
Raises:
pytest.fail: If migration message is not found in logs
"""
log_path = frontend_process._log_path
if not log_path or not os.path.exists(log_path):
pytest.fail(f"Frontend log file not found at {log_path}")
# Step 5: Kill the worker that has the request
logger.info(
f"Killing {worker_name} with PID {worker.get_pid()} processing the request"
)
terminate_process_tree(worker.get_pid(), immediate_kill=True, timeout=0)
# Step 6: Validate the completion response - should fail without migration
try:
with open(log_path, "r") as f:
log_content = f.read()
if "Stream disconnected... recreating stream..." in log_content:
logger.info(
"✓ Migration detected: Found migration message in frontend logs"
validate_completion_response(request_thread, response_list)
pytest.fail(
"Request succeeded unexpectedly when migration was disabled"
)
return
else:
except AssertionError as e:
assert "Request failed with status 500: " in str(
e
), f"Unexpected request error message: {e}"
# Step 7: Verify migration did NOT occur - should fail
try:
verify_migration_occurred(frontend)
pytest.fail(
"Expected migration did not occur - migration message not found in frontend logs"
"Migration verification unexpectedly passed when migration was disabled"
)
except Exception as e:
pytest.fail(f"Could not read frontend log file {log_path}: {e}")
except AssertionError as e:
assert "'Cannot recreate stream: ...' error found in logs" in str(
e
), f"Unexpected migration message: {e}"
@pytest.mark.vllm
@pytest.mark.gpu_1
@pytest.mark.e2e
@pytest.mark.model(FAULT_TOLERANCE_MODEL_NAME)
def test_request_migration_vllm(
def test_no_request_migration_vllm_graceful_shutdown(
request, runtime_services, predownload_models, set_ucx_tls_no_mm
):
"""
End-to-end test for worker fault tolerance with migration support.
End-to-end test for worker fault tolerance with graceful shutdown and migration disabled.
This test verifies that when a worker is killed during request processing,
the system can handle the failure gracefully and migrate the request to
another worker.
This test verifies that when migration is disabled (migration_limit=0) and a worker
receives a graceful shutdown signal (SIGTERM) during request processing, the request
fails as expected without migration. This is the opposite behavior of
test_request_migration_vllm_graceful_shutdown.
"""
# Step 1: Start the frontend
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start 2 workers sequentially
# Start worker1 first and wait for it to be ready
logger.info("Starting worker 1...")
worker1 = DynamoWorkerProcess(request, "worker1")
with worker1:
# Start worker2 after worker1 is ready
logger.info("Starting worker 2...")
worker2 = DynamoWorkerProcess(request, "worker2")
with worker2:
# Step 2: Start 2 workers sequentially with migration disabled
with DynamoWorkerProcess(request, "worker1", migration_limit=0) as worker1:
logger.info(f"Worker 1 PID: {worker1.get_pid()}")
logger.info(f"Worker 2 PID: {worker2.get_pid()}")
# Step 3: Send a test request to see which worker handles it
logger.info("Sending test request to determine worker assignment...")
test_response = send_completion_request("Who are you?", 100, timeout=60)
validate_openai_response(test_response)
logger.info("Test request completed successfully")
with DynamoWorkerProcess(request, "worker2", migration_limit=0) as worker2:
logger.info(f"Worker 2 PID: {worker2.get_pid()}")
# Step 4: Determine worker roles based on test request handling
# Frontend must use round-robin for the detection to work correctly
primary_worker, backup_worker = determine_worker_roles(worker1, worker2)
# Step 3: Send the request
request_thread, response_list = start_completion_request()
# Step 5: Send the formal request (expected to be received by the primary worker)
logger.info(
f"Sending formal request - expected to be handled by {primary_worker[1]}"
)
request_thread, response_queue = start_completion_request(
primary_worker[1]
# Step 4: Use polling to determine which worker received the request
worker, worker_name = determine_request_receiving_worker(
worker1, worker2, receiving_pattern="Decode Request ID: "
)
# Step 6: Wait 0.5 seconds after sending the formal request, then kill the primary worker
# Step 5: Gracefully shutdown the worker that has the request
logger.info(
f"Killing {primary_worker[1]} with PID {primary_worker[0].get_pid()}"
f"Gracefully shutting down {worker_name} with PID {worker.get_pid()} processing the request"
)
time.sleep(0.5)
terminate_process_tree(
primary_worker[0].get_pid(), immediate_kill=True, timeout=0
worker.get_pid(), immediate_kill=False, timeout=10
)
# Step 7: Validate the completion response
logger.info("Waiting for formal request to complete")
validate_completion_response(request_thread, response_queue)
# Step 6: Validate the completion response - should fail without migration
try:
validate_completion_response(request_thread, response_list)
pytest.fail(
"Request succeeded unexpectedly when migration was disabled"
)
except AssertionError as e:
assert "Request failed with status 500: " in str(
e
), f"Unexpected request error message: {e}"
# Step 8: Verify migration occurred
logger.info("Checking for migration message in frontend logs")
# Step 7: Verify migration did NOT occur - should fail
try:
verify_migration_occurred(frontend)
logger.info(
"Test completed successfully - migration is detected and the request was successful"
pytest.fail(
"Migration verification unexpectedly passed when migration was disabled"
)
except AssertionError as e:
assert "'Cannot recreate stream: ...' error found in logs" in str(
e
), f"Unexpected migration message: {e}"
tests/fault_tolerance/migration/utils.py 0 → 100644
View file @ 59535682
# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
import logging
import shutil
import threading
import time
import pytest
import requests
from tests.utils.constants import FAULT_TOLERANCE_MODEL_NAME
from tests.utils.engine_process import FRONTEND_PORT
from tests.utils.managed_process import ManagedProcess
logger = logging.getLogger(__name__)
class DynamoFrontendProcess(ManagedProcess):
"""Process manager for Dynamo frontend"""
def __init__(self, request):
command = ["python", "-m", "dynamo.frontend", "--router-mode", "round-robin"]
log_dir = f"{request.node.name}_frontend"
# Clean up any existing log directory from previous runs
try:
shutil.rmtree(log_dir)
logger.info(f"Cleaned up existing log directory: {log_dir}")
except FileNotFoundError:
# Directory doesn't exist, which is fine
pass
super().__init__(
command=command,
display_output=True,
terminate_existing=True,
log_dir=log_dir,
)
def start_completion_request() -> tuple:
"""
Start a long-running completion request in a separate thread.
Returns:
tuple: (request_thread, response_list)
"""
response_list = [] # Thread safe is not required as only one thread writes to it
def send_request():
prompt = "Tell me a long long long story about yourself?"
max_tokens = 8192
timeout = 240 # Extended timeout for long request
payload = {
"model": FAULT_TOLERANCE_MODEL_NAME,
"prompt": prompt,
"max_tokens": max_tokens,
}
headers = {"Content-Type": "application/json"}
logger.info(
f"Sending completion request with prompt: '{prompt[:50]}...' and max_tokens: {max_tokens}"
)
try:
response = requests.post(
f"http://localhost:{FRONTEND_PORT}/v1/completions",
headers=headers,
json=payload,
timeout=timeout,
)
logger.info(f"Received response with status code: {response.status_code}")
response_list.append(response)
except Exception as e:
logger.error(f"Request failed with error: {e}")
request_thread = threading.Thread(target=send_request, daemon=True)
request_thread.start()
return request_thread, response_list
def determine_request_receiving_worker(
worker1: ManagedProcess, worker2: ManagedProcess, receiving_pattern: str
) -> tuple:
"""
Determine which worker received the request using parallel polling.
Args:
worker1: First worker process
worker2: Second worker process
receiving_pattern: Log pattern indicating request receipt
Returns:
Tuple of (worker_with_request, name_of_worker_with_request)
"""
worker1_results: list[bool] = []
worker2_results: list[bool] = []
# Poll both workers in parallel
def poll_worker(worker: ManagedProcess, result_list: list[bool]):
max_wait_ms = 500
poll_interval_ms = 5
max_iterations = max_wait_ms // poll_interval_ms
iteration = 0
while iteration < max_iterations:
# Check if the worker logs contain 'New Request ID:' message
try:
with open(worker.log_path, "r") as f:
log_content = f.read()
if receiving_pattern in log_content:
result_list.append(True)
return
except Exception as e:
logger.error(f"Could not read log file {worker.log_path}: {e}")
return
time.sleep(poll_interval_ms / 1000.0)
iteration += 1
# Look for which worker received the request
thread1 = threading.Thread(
target=poll_worker, args=(worker1, worker1_results), daemon=True
)
thread2 = threading.Thread(
target=poll_worker, args=(worker2, worker2_results), daemon=True
)
thread1.start()
thread2.start()
thread1.join(timeout=1)
thread2.join(timeout=1)
# Get results from lists
worker1_received = worker1_results[0] if worker1_results else False
worker2_received = worker2_results[0] if worker2_results else False
if worker1_received and not worker2_received:
logger.info("Request was received by Worker 1")
return worker1, "Worker 1"
elif worker2_received and not worker1_received:
logger.info("Request was received by Worker 2")
return worker2, "Worker 2"
elif worker1_received and worker2_received:
pytest.fail("Both workers received the request")
else:
pytest.fail("Neither worker received the request")
def validate_completion_response(
request_thread: threading.Thread, response_list: list
) -> None:
"""
Wait for and validate the completion response after worker failure.
Args:
request_thread: The thread running the completion request
response_list: List containing the response from the request
"""
request_thread.join(timeout=240)
if request_thread.is_alive():
pytest.fail("Request did not complete within 240 seconds")
# Get the response
if len(response_list) != 1:
pytest.fail(f"Received {len(response_list)} responses, expected 1")
response = response_list[0]
assert (
response.status_code == 200
), f"Request failed with status {response.status_code}: {response.text}"
try:
data = response.json()
except ValueError:
pytest.fail(f"Response is not valid JSON: {response.text}")
# Validate OpenAI completion response structure
assert "choices" in data, f"Response missing 'choices' field: {data}"
assert len(data["choices"]) > 0, f"Response has empty 'choices': {data}"
assert "text" in data["choices"][0], f"Response choice missing 'text' field: {data}"
assert data["choices"][0]["text"], f"Response text is empty: {data}"
logger.info(
f"Received valid completion response: {data['choices'][0]['text'][:100]}..."
)
logger.info("Request completed successfully")
def verify_migration_occurred(frontend_process: DynamoFrontendProcess) -> None:
"""
Verify that migration occurred by checking frontend logs for stream disconnection message.
Args:
frontend_process: The frontend process to check logs for
"""
log_path = frontend_process.log_path
try:
with open(log_path, "r") as f:
log_content = f.read()
except Exception as e:
pytest.fail(f"Could not read frontend log file {log_path}: {e}")
assert (
"Stream disconnected... recreating stream..." in log_content
), "'Stream disconnected... recreating stream...' message not found in logs"
assert (
"Cannot recreate stream: " not in log_content
), "'Cannot recreate stream: ...' error found in logs"
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment