wehub-resource-sync
1660 lines
54 KiB
Python
1660 lines
54 KiB
Python
"""
|
|
Some instructions on writing CLI tests:
|
|
1. Look at test_ray_start for a simple output test example.
|
|
2. To get a valid regex, start with copy-pasting your output from a captured
|
|
version (no formatting). Then escape ALL regex characters (parenthesis,
|
|
brackets, dots, etc.). THEN add ".+" to all the places where info might
|
|
change run to run.
|
|
3. Look at test_ray_up for an example of how to mock AWS, commands,
|
|
and autoscaler config.
|
|
4. Print your outputs!!!! Tests are impossible to debug if they fail
|
|
and you did not print anything. Since command output is captured by click,
|
|
MAKE SURE YOU print(result.output) when tests fail!!!
|
|
|
|
WARNING: IF YOU MOCK AWS, DON'T FORGET THE AWS_CREDENTIALS FIXTURE.
|
|
THIS IS REQUIRED SO BOTO3 DOES NOT ACCESS THE ACTUAL AWS SERVERS.
|
|
|
|
Note: config cache does not work with AWS mocks since the AWS resource ids are
|
|
randomized each time.
|
|
"""
|
|
import glob
|
|
import json
|
|
import multiprocessing as mp
|
|
import multiprocessing.connection
|
|
import os
|
|
import re
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from contextlib import contextmanager
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
from unittest import mock
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
import yaml
|
|
from click.testing import CliRunner
|
|
from moto import mock_aws
|
|
from testfixtures import Replacer
|
|
from testfixtures.popen import MockPopen, PopenBehaviour
|
|
|
|
import ray
|
|
import ray._private.ray_constants as ray_constants
|
|
import ray.autoscaler._private.aws.config as aws_config
|
|
import ray.autoscaler._private.constants as autoscaler_constants
|
|
import ray.scripts.scripts as scripts
|
|
import ray.tests.aws.utils.helpers as aws_helpers
|
|
from ray._common.network_utils import build_address, parse_address
|
|
from ray._common.test_utils import wait_for_condition
|
|
from ray._common.utils import get_default_ray_temp_dir
|
|
from ray.autoscaler._private.providers import _get_node_provider
|
|
from ray.autoscaler._private.util import prepare_config
|
|
from ray.cluster_utils import cluster_not_supported
|
|
from ray.util.check_open_ports import check_open_ports
|
|
from ray.util.state import list_nodes
|
|
|
|
import psutil
|
|
|
|
boto3_list = [
|
|
{
|
|
"InstanceType": "t1.micro",
|
|
"VCpuInfo": {"DefaultVCpus": 1},
|
|
"MemoryInfo": {"SizeInMiB": 627},
|
|
},
|
|
{
|
|
"InstanceType": "t3a.small",
|
|
"VCpuInfo": {"DefaultVCpus": 2},
|
|
"MemoryInfo": {"SizeInMiB": 2048},
|
|
},
|
|
{
|
|
"InstanceType": "m4.4xlarge",
|
|
"VCpuInfo": {"DefaultVCpus": 16},
|
|
"MemoryInfo": {"SizeInMiB": 65536},
|
|
},
|
|
{
|
|
"InstanceType": "p3.8xlarge",
|
|
"VCpuInfo": {"DefaultVCpus": 32},
|
|
"MemoryInfo": {"SizeInMiB": 249856},
|
|
"GpuInfo": {"Gpus": [{"Name": "V100", "Count": 4}]},
|
|
},
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def configure_lang():
|
|
"""Configure output for travis + click."""
|
|
if sys.platform != "darwin":
|
|
os.environ["LC_ALL"] = "C.UTF-8"
|
|
os.environ["LANG"] = "C.UTF-8"
|
|
|
|
|
|
@pytest.fixture
|
|
def cleanup_ray():
|
|
"""Shutdown all ray instances"""
|
|
yield
|
|
runner = CliRunner()
|
|
runner.invoke(scripts.stop, ["--force"])
|
|
ray.shutdown()
|
|
|
|
|
|
@pytest.fixture
|
|
def configure_aws():
|
|
"""Mocked AWS Credentials for moto."""
|
|
os.environ["LC_ALL"] = "C.UTF-8"
|
|
os.environ["LANG"] = "C.UTF-8"
|
|
os.environ["AWS_ACCESS_KEY_ID"] = "testing"
|
|
os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
|
|
os.environ["AWS_SECURITY_TOKEN"] = "testing"
|
|
os.environ["AWS_SESSION_TOKEN"] = "testing"
|
|
|
|
# moto (boto3 mock) only allows a hardcoded set of AMIs
|
|
# Use mock_aws context manager and boto3 to find the AMI
|
|
import boto3
|
|
|
|
# In moto 5.x, AWS managed policies (e.g., AmazonEC2FullAccess) are not
|
|
# loaded by default for performance. Enable them since the autoscaler
|
|
# attaches these policies to the IAM role.
|
|
with mock_aws(config={"iam": {"load_aws_managed_policies": True}}):
|
|
ec2_client = boto3.client("ec2", region_name="us-west-2")
|
|
images = ec2_client.describe_images(
|
|
Filters=[{"Name": "name", "Values": ["Deep Learning AMI Ubuntu*"]}]
|
|
)["Images"]
|
|
dlami = images[0]["ImageId"]
|
|
aws_config.DEFAULT_AMI["us-west-2"] = dlami
|
|
list_instances_mock = MagicMock(return_value=boto3_list)
|
|
with patch(
|
|
"ray.autoscaler._private.aws.node_provider.list_ec2_instances",
|
|
list_instances_mock,
|
|
):
|
|
yield
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def _unlink_test_ssh_key():
|
|
"""Use this to remove the keys spawned by ray up."""
|
|
yield
|
|
try:
|
|
for path in glob.glob(os.path.expanduser("~/.ssh/__test-cli_key*")):
|
|
os.remove(path)
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
|
|
def _start_ray_and_block(
|
|
runner, child_conn: multiprocessing.connection.Connection, as_head: bool
|
|
):
|
|
"""Utility function to start a CLI command with `ray start --block`
|
|
|
|
This function is expected to be run in another process, where `child_conn` is used
|
|
for IPC with the parent.
|
|
"""
|
|
args = ["--block"]
|
|
if as_head:
|
|
args.append("--head")
|
|
else:
|
|
# Worker node
|
|
args.append(f"--address=localhost:{ray_constants.DEFAULT_PORT}")
|
|
|
|
result = runner.invoke(
|
|
scripts.start,
|
|
args,
|
|
)
|
|
|
|
# Should be blocked until stopped by signals (SIGTERM)
|
|
child_conn.send(result.output)
|
|
|
|
|
|
def _debug_die(output, assert_msg: str = ""):
|
|
print("!!!!")
|
|
print(output)
|
|
print("!!!!")
|
|
assert False, assert_msg
|
|
|
|
|
|
def _fail_if_false(
|
|
predicate: bool, stdout: Optional[str] = "", assert_msg: Optional[str] = ""
|
|
):
|
|
if not predicate:
|
|
_debug_die(stdout, assert_msg)
|
|
|
|
|
|
def _die_on_error(result):
|
|
_fail_if_false(result.exit_code == 0, result.output)
|
|
|
|
|
|
def test_log_unexpected_subprocess_exit_details(monkeypatch, tmp_path):
|
|
logs_dir = tmp_path
|
|
(logs_dir / "gcs_server.err").write_text("sentinel_tail_line\n", encoding="utf-8")
|
|
|
|
emitted_errors = []
|
|
|
|
class _IndentedContext:
|
|
def __enter__(self):
|
|
return None
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def _render_message(message, args):
|
|
try:
|
|
return str(message).format(*args)
|
|
except Exception:
|
|
return f"{message} {' '.join(map(str, args))}"
|
|
|
|
def _capture_error(message, *args, **kwargs):
|
|
emitted_errors.append(_render_message(message, args))
|
|
|
|
monkeypatch.setattr(scripts.cli_logger, "indented", lambda: _IndentedContext())
|
|
monkeypatch.setattr(scripts.cli_logger, "error", _capture_error)
|
|
monkeypatch.setattr(scripts.cli_logger, "warning", lambda *args, **kwargs: None)
|
|
monkeypatch.setattr(scripts.cli_logger, "newline", lambda *args, **kwargs: None)
|
|
|
|
process_exit_logger = MagicMock()
|
|
fake_process = MagicMock(returncode=137)
|
|
|
|
scripts._log_unexpected_subprocess_exit_details(
|
|
[("gcs_server", fake_process)],
|
|
str(logs_dir),
|
|
process_exit_logger,
|
|
)
|
|
|
|
process_exit_logger.error.assert_called_once()
|
|
logged_message = process_exit_logger.error.call_args.args[1]
|
|
assert "Some Ray subprocesses exited unexpectedly:" in logged_message
|
|
assert "gcs_server [exit code=" in logged_message
|
|
assert any("BEGIN gcs_server.err tail (exit code(s)=" in x for x in emitted_errors)
|
|
assert any("sentinel_tail_line" in x for x in emitted_errors)
|
|
|
|
|
|
def _debug_check_line_by_line(result, expected_lines):
|
|
"""Print the result and expected output line-by-line."""
|
|
output_lines = result.output.split("\n")
|
|
i = 0
|
|
|
|
for out in output_lines:
|
|
if i >= len(expected_lines):
|
|
i += 1
|
|
print("!!!!!! Expected fewer lines")
|
|
context = [f"CONTEXT: {line}" for line in output_lines[i - 3 : i]]
|
|
print("\n".join(context))
|
|
extra = [f"-- {line}" for line in output_lines[i:]]
|
|
print("\n".join(extra))
|
|
break
|
|
|
|
exp = expected_lines[i]
|
|
matched = re.fullmatch(exp + r" *", out) is not None
|
|
if not matched:
|
|
print(f"{i:>3}: {out}")
|
|
print(f"!!! ^ ERROR: Expected (regex): {repr(exp)}")
|
|
else:
|
|
print(f"{i:>3}: {out}")
|
|
i += 1
|
|
if i < len(expected_lines):
|
|
print("!!! ERROR: Expected extra lines (regex):")
|
|
for line in expected_lines[i:]:
|
|
print(repr(line))
|
|
|
|
assert False, (result.output, expected_lines)
|
|
|
|
|
|
@contextmanager
|
|
def _setup_popen_mock(commands_mock, commands_verifier=None):
|
|
"""
|
|
Mock subprocess.Popen's behavior and if applicable, intercept the commands
|
|
received by Popen and check if they are as expected using
|
|
commands_verifier provided by caller.
|
|
TODO(xwjiang): Ideally we should write a lexical analyzer that can parse
|
|
in a more intelligent way.
|
|
"""
|
|
Popen = MockPopen()
|
|
Popen.set_default(behaviour=commands_mock)
|
|
|
|
with Replacer() as replacer:
|
|
replacer.replace("subprocess.Popen", Popen)
|
|
yield
|
|
|
|
if commands_verifier:
|
|
assert commands_verifier(Popen.all_calls)
|
|
|
|
|
|
def _load_output_pattern(name):
|
|
pattern_dir = Path(__file__).parent / "test_cli_patterns"
|
|
with open(str(pattern_dir / name)) as f:
|
|
# Remove \n from each line.
|
|
# Substitute the Ray version in each line containing the string
|
|
# {ray_version}.
|
|
out = []
|
|
for x in f.readlines():
|
|
if "{ray_version}" in x:
|
|
out.append(x[:-1].format(ray_version=ray.__version__))
|
|
else:
|
|
out.append(x[:-1])
|
|
return out
|
|
|
|
|
|
def _check_output_via_pattern(name, result):
|
|
expected_lines = _load_output_pattern(name)
|
|
|
|
if result.exception is not None:
|
|
raise result.exception from None
|
|
print(result.output)
|
|
expected = r" *\n".join(expected_lines) + "\n?"
|
|
if re.fullmatch(expected, result.output) is None:
|
|
_debug_check_line_by_line(result, expected_lines)
|
|
|
|
assert result.exit_code == 0
|
|
|
|
|
|
DEFAULT_TEST_CONFIG_PATH = str(
|
|
Path(__file__).parent / "test_cli_patterns" / "test_ray_up_config.yaml"
|
|
)
|
|
|
|
MISSING_MAX_WORKER_CONFIG_PATH = str(
|
|
Path(__file__).parent
|
|
/ "test_cli_patterns"
|
|
/ "test_ray_up_no_max_worker_config.yaml"
|
|
)
|
|
|
|
DOCKER_TEST_CONFIG_PATH = str(
|
|
Path(__file__).parent / "test_cli_patterns" / "test_ray_up_docker_config.yaml"
|
|
)
|
|
|
|
|
|
def test_enable_usage_stats(monkeypatch, tmp_path):
|
|
tmp_usage_stats_config_path = tmp_path / "config.json"
|
|
monkeypatch.setenv("RAY_USAGE_STATS_CONFIG_PATH", str(tmp_usage_stats_config_path))
|
|
runner = CliRunner()
|
|
runner.invoke(scripts.enable_usage_stats, [])
|
|
assert '{"usage_stats": true}' == tmp_usage_stats_config_path.read_text()
|
|
|
|
|
|
def test_disable_usage_stats(monkeypatch, tmp_path):
|
|
tmp_usage_stats_config_path = tmp_path / "config.json"
|
|
monkeypatch.setenv("RAY_USAGE_STATS_CONFIG_PATH", str(tmp_usage_stats_config_path))
|
|
runner = CliRunner()
|
|
runner.invoke(scripts.disable_usage_stats, [])
|
|
assert '{"usage_stats": false}' == tmp_usage_stats_config_path.read_text()
|
|
|
|
|
|
# We add`env={"RAY_USAGE_STATS_PROMPT_ENABLED": "0"}` in these tests because
|
|
# it seems like interactive terminal detection works differently in python 3.8
|
|
# compared to 3.7. Without this, tests would fail.
|
|
# Todo: This should be removed again. Also, some tests are currently skipped.
|
|
@pytest.mark.skipif(sys.platform == "darwin", reason="Currently failing on OSX")
|
|
def test_ray_start(configure_lang, monkeypatch, tmp_path, cleanup_ray):
|
|
monkeypatch.setenv("RAY_USAGE_STATS_CONFIG_PATH", str(tmp_path / "config.json"))
|
|
runner = CliRunner(env={"RAY_USAGE_STATS_PROMPT_ENABLED": "0"})
|
|
temp_dir = os.path.join("/tmp", uuid.uuid4().hex)
|
|
result = runner.invoke(
|
|
scripts.start,
|
|
[
|
|
"--head",
|
|
"--log-style=pretty",
|
|
"--log-color",
|
|
"False",
|
|
"--port",
|
|
"0",
|
|
"--temp-dir",
|
|
temp_dir,
|
|
],
|
|
)
|
|
|
|
# Check that --temp-dir arg worked:
|
|
assert os.path.isfile(os.path.join(temp_dir, "ray_current_cluster"))
|
|
assert os.path.isdir(os.path.join(temp_dir, "session_latest"))
|
|
_die_on_error(result)
|
|
|
|
_die_on_error(runner.invoke(scripts.stop))
|
|
|
|
if ray_constants.IS_WINDOWS_OR_OSX:
|
|
_check_output_via_pattern("test_ray_start_windows_osx.txt", result)
|
|
else:
|
|
_check_output_via_pattern("test_ray_start.txt", result)
|
|
|
|
# Check that we can rerun `ray start` even though the cluster address file
|
|
# is already written.
|
|
_die_on_error(
|
|
runner.invoke(
|
|
scripts.start,
|
|
[
|
|
"--head",
|
|
"--log-style=pretty",
|
|
"--log-color",
|
|
"False",
|
|
"--port",
|
|
"0",
|
|
"--temp-dir",
|
|
temp_dir,
|
|
],
|
|
)
|
|
)
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
sys.platform == "darwin" and "travis" in os.environ.get("USER", ""),
|
|
reason=("Mac builds don't provide proper locale support"),
|
|
)
|
|
def test_ray_start_worker_can_specify_temp_dir(configure_lang, tmp_path, cleanup_ray):
|
|
"""
|
|
Verify that ray start --temp-dir works on worker nodes independently of head node.
|
|
"""
|
|
runner = CliRunner(env={"RAY_USAGE_STATS_PROMPT_ENABLED": "0"})
|
|
|
|
# Start head node without specifying temp-dir
|
|
result = runner.invoke(
|
|
scripts.start,
|
|
[
|
|
"--head",
|
|
],
|
|
)
|
|
print(result.output)
|
|
_die_on_error(result)
|
|
|
|
# Start worker node with temp-dir specified
|
|
worker_temp_dir = os.path.join("/tmp", uuid.uuid4().hex)
|
|
result = runner.invoke(
|
|
scripts.start,
|
|
[
|
|
f"--address=localhost:{ray_constants.DEFAULT_PORT}",
|
|
f"--temp-dir={worker_temp_dir}",
|
|
],
|
|
)
|
|
_die_on_error(result)
|
|
|
|
# Check that worker temp-dir was created at the specified location
|
|
assert os.path.isfile(os.path.join(worker_temp_dir, "ray_current_cluster"))
|
|
assert os.path.isdir(os.path.join(worker_temp_dir, "session_latest"))
|
|
|
|
# Check that we can rerun `ray start` even though the cluster address file
|
|
# is already written.
|
|
_die_on_error(
|
|
runner.invoke(
|
|
scripts.start,
|
|
[
|
|
f"--address=localhost:{ray_constants.DEFAULT_PORT}",
|
|
f"--temp-dir={worker_temp_dir}",
|
|
],
|
|
)
|
|
)
|
|
|
|
|
|
def _ray_start_hook(ray_params, head):
|
|
os.makedirs(ray_params.temp_dir, exist_ok=True)
|
|
with open(os.path.join(ray_params.temp_dir, "ray_hook_ok"), "w") as f:
|
|
f.write("HOOK_OK")
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
sys.platform == "darwin" and "travis" in os.environ.get("USER", ""),
|
|
reason=("Mac builds don't provide proper locale support"),
|
|
)
|
|
def test_ray_start_hook(configure_lang, monkeypatch, cleanup_ray):
|
|
monkeypatch.setenv("RAY_START_HOOK", "ray.tests.test_cli._ray_start_hook")
|
|
runner = CliRunner(env={"RAY_USAGE_STATS_PROMPT_ENABLED": "0"})
|
|
temp_dir = os.path.join("/tmp", uuid.uuid4().hex)
|
|
runner.invoke(
|
|
scripts.start,
|
|
[
|
|
"--head",
|
|
"--log-style=pretty",
|
|
"--log-color",
|
|
"False",
|
|
"--port",
|
|
"0",
|
|
"--temp-dir",
|
|
temp_dir,
|
|
],
|
|
)
|
|
|
|
# Check that the hook executed.
|
|
assert os.path.exists(temp_dir)
|
|
assert os.path.exists(os.path.join(temp_dir, "ray_hook_ok"))
|
|
|
|
_die_on_error(runner.invoke(scripts.stop))
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
sys.version_info.minor >= 8, reason="Currently fails with Python 3.8+"
|
|
)
|
|
@pytest.mark.skipif(
|
|
sys.platform == "darwin",
|
|
reason=("Mac builds don't provide proper locale support. "),
|
|
)
|
|
@pytest.mark.skipif(
|
|
sys.platform == "win32", reason="Windows signal handling not compatible"
|
|
)
|
|
def test_ray_start_head_block_and_signals(
|
|
configure_lang, monkeypatch, tmp_path, cleanup_ray
|
|
):
|
|
"""Test `ray start` with `--block` as heads and workers and signal handles"""
|
|
|
|
monkeypatch.setenv("RAY_USAGE_STATS_CONFIG_PATH", str(tmp_path / "config.json"))
|
|
runner = CliRunner(env={"RAY_USAGE_STATS_PROMPT_ENABLED": "0"})
|
|
|
|
head_parent_conn, head_child_conn = mp.Pipe()
|
|
|
|
# Run `ray start --block --head` in another process and blocks
|
|
head_proc = mp.Process(
|
|
target=_start_ray_and_block,
|
|
kwargs={"runner": runner, "child_conn": head_child_conn, "as_head": True},
|
|
)
|
|
head_proc._start_method = "spawn"
|
|
|
|
# Run
|
|
head_proc.start()
|
|
|
|
# Give it some time to start various subprocesses and `ray stop`
|
|
# A smaller interval seems to cause occasional failure as the head process
|
|
# was stopped too early before spawning all the subprocesses.
|
|
time.sleep(5)
|
|
|
|
# Terminate some of the children process
|
|
children = psutil.Process(head_proc.pid).children()
|
|
|
|
# Terminate everyone other than GCS
|
|
# NOTE(rickyyx): The choice of picking GCS is arbitrary.
|
|
gcs_proc = None
|
|
for child in children:
|
|
if "gcs_server" in child.name():
|
|
gcs_proc = child
|
|
continue
|
|
child.terminate()
|
|
child.wait(5)
|
|
|
|
if not head_proc.is_alive():
|
|
# NOTE(rickyyx): call recv() here is safe since the process
|
|
# is guaranteed to be terminated.
|
|
_fail_if_false(
|
|
False,
|
|
head_parent_conn.recv(),
|
|
(
|
|
"`ray start --head --block` should not exit"
|
|
f"({head_proc.exitcode}) when a subprocess is "
|
|
"terminated with SIGTERM."
|
|
),
|
|
)
|
|
|
|
# Kill the GCS last should unblock the CLI
|
|
if gcs_proc:
|
|
gcs_proc.kill()
|
|
gcs_proc.wait(10)
|
|
|
|
# NOTE(rickyyx): The wait here is needed for the `head_proc`
|
|
# process to exit
|
|
head_proc.join(5)
|
|
|
|
exit_log = Path(
|
|
os.path.join(
|
|
get_default_ray_temp_dir(),
|
|
ray_constants.SESSION_LATEST,
|
|
"logs",
|
|
"ray_process_exit.log",
|
|
)
|
|
)
|
|
assert exit_log.exists(), f"ray_process_exit.log not found at {exit_log}"
|
|
content = exit_log.read_text(encoding="utf-8")
|
|
assert (
|
|
"Some Ray subprocesses exited unexpectedly:" in content
|
|
), "Expected message not found in ray_process_exit.log."
|
|
|
|
# Process with "--block" should be dead with a subprocess killed
|
|
if head_proc.is_alive() or head_proc.exitcode == 0:
|
|
# NOTE(rickyyx): call recv() here is safe since the process
|
|
# is guaranteed to be terminated thus invocation is non-blocking.
|
|
_fail_if_false(
|
|
False,
|
|
head_parent_conn.recv() if not head_proc.is_alive() else "still alive",
|
|
(
|
|
"Head process should have exited with errors when one of"
|
|
f" subprocesses killed. But exited={head_proc.exitcode}"
|
|
),
|
|
)
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
sys.version_info.minor >= 8, reason="Currently fails with Python 3.8+"
|
|
)
|
|
@pytest.mark.skipif(
|
|
sys.platform == "darwin",
|
|
reason=("Mac builds don't provide proper locale support. "),
|
|
)
|
|
@pytest.mark.skipif(
|
|
sys.platform == "win32", reason="Windows signal handling not compatible"
|
|
)
|
|
def test_ray_start_block_and_stop(configure_lang, monkeypatch, tmp_path, cleanup_ray):
|
|
"""Test `ray start` with `--block` as heads and workers and `ray stop`"""
|
|
monkeypatch.setenv("RAY_USAGE_STATS_CONFIG_PATH", str(tmp_path / "config.json"))
|
|
runner = CliRunner(env={"RAY_USAGE_STATS_PROMPT_ENABLED": "0"})
|
|
|
|
head_parent_conn, head_child_conn = mp.Pipe()
|
|
worker_parent_conn, worker_child_conn = mp.Pipe()
|
|
|
|
# Run `ray start --block --head` in another process and blocks
|
|
head_proc = mp.Process(
|
|
target=_start_ray_and_block,
|
|
kwargs={"runner": runner, "child_conn": head_child_conn, "as_head": True},
|
|
)
|
|
head_proc._start_method = "spawn"
|
|
|
|
# Run `ray start --block --address=localhost:DEFAULT_PORT`
|
|
worker_proc = mp.Process(
|
|
target=_start_ray_and_block,
|
|
kwargs={"runner": runner, "child_conn": worker_child_conn, "as_head": False},
|
|
)
|
|
worker_proc._start_method = "spawn"
|
|
|
|
try:
|
|
# Run
|
|
head_proc.start()
|
|
worker_proc.start()
|
|
print("head pid ", head_proc.pid)
|
|
print("worker pid: ", worker_proc.pid)
|
|
|
|
# Without sleep list_nodes sometimes failed.
|
|
time.sleep(5)
|
|
# Wait until all nodes are registered and started.
|
|
wait_for_condition(lambda: len(list_nodes()) == 2)
|
|
# When ray start --block is called, before it blocks,
|
|
# it pings GCS to get the Ray & Python version. If GCS
|
|
# is killed before that by ray stop, it can trigger GCS
|
|
# timeout error which can break `ray start --block`.
|
|
# To avoid this issue, we sleep enough before we stop ray.
|
|
time.sleep(5)
|
|
|
|
stop_result = runner.invoke(scripts.stop)
|
|
print("result, ", stop_result.output)
|
|
print("exit code, ", stop_result.exit_code)
|
|
_die_on_error(stop_result)
|
|
|
|
# Process with "--block" should be blocked forever w/o
|
|
# termination by signals
|
|
if not head_proc.is_alive():
|
|
# NOTE(rickyyx): call recv() here is safe since the process
|
|
# is guaranteed to be terminated.
|
|
if head_proc.exitcode == 1:
|
|
assert False, (
|
|
"ray start --head --block is failed "
|
|
"due to unexpected component failures."
|
|
)
|
|
_fail_if_false(
|
|
False,
|
|
head_parent_conn.recv(),
|
|
(
|
|
"`ray start --head --block` (head) should block forever even"
|
|
" though Ray subprocesses are stopped normally. But "
|
|
f"it exited with {head_proc.exitcode} early. \n"
|
|
f"Stop command: {stop_result.output}"
|
|
),
|
|
)
|
|
|
|
if not worker_proc.is_alive():
|
|
if worker_proc.exitcode == 1:
|
|
assert False, (
|
|
"ray start --address=<head_ip> --block` (worker) is failed "
|
|
"due to unexpected component failures."
|
|
)
|
|
_fail_if_false(
|
|
False,
|
|
worker_parent_conn.recv(),
|
|
(
|
|
"`ray start --block` should block forever even"
|
|
" though Ray subprocesses are stopped normally. But"
|
|
f"it exited with {worker_proc.exitcode} already. \n"
|
|
f"Stop command: {stop_result.output}"
|
|
),
|
|
)
|
|
|
|
# Stop both worker and head with SIGTERM
|
|
head_proc.terminate()
|
|
worker_proc.terminate()
|
|
|
|
head_proc.join(15)
|
|
worker_proc.join(15)
|
|
|
|
worker_output = "cannot poll"
|
|
head_output = "cannot poll"
|
|
if head_parent_conn.poll(5):
|
|
head_output = head_parent_conn.recv()
|
|
if worker_parent_conn.poll(5):
|
|
worker_output = worker_parent_conn.recv()
|
|
|
|
print("head ", head_output)
|
|
print("worker ", worker_output)
|
|
assert not head_proc.is_alive(), "head node is not killed."
|
|
assert not worker_proc.is_alive(), "worker node is not killed."
|
|
|
|
_fail_if_false(
|
|
head_proc.exitcode == 0,
|
|
head_output,
|
|
f"Head process failed unexpectedly({head_proc.exitcode})",
|
|
)
|
|
_fail_if_false(
|
|
worker_proc.exitcode == 0,
|
|
worker_output,
|
|
f"Worker process failed unexpectedly({worker_proc.exitcode})",
|
|
)
|
|
finally:
|
|
head_proc.kill()
|
|
worker_proc.kill()
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
sys.platform == "darwin" and "travis" in os.environ.get("USER", ""),
|
|
reason=("Mac builds don't provide proper locale support"),
|
|
)
|
|
def test_ray_up(
|
|
configure_lang, _unlink_test_ssh_key, configure_aws, monkeypatch, tmp_path
|
|
):
|
|
monkeypatch.setenv("RAY_USAGE_STATS_CONFIG_PATH", str(tmp_path / "config.json"))
|
|
|
|
def commands_mock(command, stdin):
|
|
# if we want to have e.g. some commands fail,
|
|
# we can have overrides happen here.
|
|
# unfortunately, cutting out SSH prefixes and such
|
|
# is, to put it lightly, non-trivial
|
|
if "uptime" in command:
|
|
return PopenBehaviour(stdout=b"MOCKED uptime")
|
|
if "rsync" in command:
|
|
return PopenBehaviour(stdout=b"MOCKED rsync")
|
|
if "ray" in command:
|
|
return PopenBehaviour(stdout=b"MOCKED ray")
|
|
return PopenBehaviour(stdout=b"MOCKED GENERIC")
|
|
|
|
with _setup_popen_mock(commands_mock):
|
|
# config cache does not work with mocks
|
|
runner = CliRunner(env={"RAY_USAGE_STATS_PROMPT_ENABLED": "0"})
|
|
result = runner.invoke(
|
|
scripts.up,
|
|
[
|
|
DEFAULT_TEST_CONFIG_PATH,
|
|
"--no-config-cache",
|
|
"-y",
|
|
"--log-style=pretty",
|
|
"--log-color",
|
|
"False",
|
|
],
|
|
)
|
|
_check_output_via_pattern("test_ray_up.txt", result)
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
sys.platform == "darwin" and "travis" in os.environ.get("USER", ""),
|
|
reason=("Mac builds don't provide proper locale support"),
|
|
)
|
|
def test_ray_up_docker(
|
|
configure_lang, _unlink_test_ssh_key, configure_aws, monkeypatch, tmp_path
|
|
):
|
|
monkeypatch.setenv("RAY_USAGE_STATS_CONFIG_PATH", str(tmp_path / "config.json"))
|
|
|
|
def commands_mock(command, stdin):
|
|
# if we want to have e.g. some commands fail,
|
|
# we can have overrides happen here.
|
|
# unfortunately, cutting out SSH prefixes and such
|
|
# is, to put it lightly, non-trivial
|
|
if ".Config.Env" in command:
|
|
return PopenBehaviour(stdout=b"{}")
|
|
if "uptime" in command:
|
|
return PopenBehaviour(stdout=b"MOCKED uptime")
|
|
if "rsync" in command:
|
|
return PopenBehaviour(stdout=b"MOCKED rsync")
|
|
if "ray" in command:
|
|
return PopenBehaviour(stdout=b"MOCKED ray")
|
|
return PopenBehaviour(stdout=b"MOCKED GENERIC")
|
|
|
|
with _setup_popen_mock(commands_mock):
|
|
# config cache does not work with mocks
|
|
runner = CliRunner(env={"RAY_USAGE_STATS_PROMPT_ENABLED": "0"})
|
|
result = runner.invoke(
|
|
scripts.up,
|
|
[
|
|
DOCKER_TEST_CONFIG_PATH,
|
|
"--no-config-cache",
|
|
"-y",
|
|
"--log-style=pretty",
|
|
"--log-color",
|
|
"False",
|
|
],
|
|
)
|
|
_check_output_via_pattern("test_ray_up_docker.txt", result)
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
sys.platform == "darwin" and "travis" in os.environ.get("USER", ""),
|
|
reason=("Mac builds don't provide proper locale support"),
|
|
)
|
|
def test_ray_up_record(
|
|
configure_lang, _unlink_test_ssh_key, configure_aws, monkeypatch, tmp_path
|
|
):
|
|
monkeypatch.setenv("RAY_USAGE_STATS_CONFIG_PATH", str(tmp_path / "config.json"))
|
|
|
|
def commands_mock(command, stdin):
|
|
# if we want to have e.g. some commands fail,
|
|
# we can have overrides happen here.
|
|
# unfortunately, cutting out SSH prefixes and such
|
|
# is, to put it lightly, non-trivial
|
|
if "uptime" in command:
|
|
return PopenBehaviour(stdout=b"MOCKED uptime")
|
|
if "rsync" in command:
|
|
return PopenBehaviour(stdout=b"MOCKED rsync")
|
|
if "ray" in command:
|
|
return PopenBehaviour(stdout=b"MOCKED ray")
|
|
return PopenBehaviour(stdout=b"MOCKED GENERIC")
|
|
|
|
with _setup_popen_mock(commands_mock):
|
|
# config cache does not work with mocks
|
|
runner = CliRunner(env={"RAY_USAGE_STATS_PROMPT_ENABLED": "0"})
|
|
result = runner.invoke(
|
|
scripts.up,
|
|
[DEFAULT_TEST_CONFIG_PATH, "--no-config-cache", "-y", "--log-style=record"],
|
|
)
|
|
_check_output_via_pattern("test_ray_up_record.txt", result)
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
sys.platform == "darwin" and "travis" in os.environ.get("USER", ""),
|
|
reason=("Mac builds don't provide proper locale support"),
|
|
)
|
|
def test_ray_attach(configure_lang, configure_aws, _unlink_test_ssh_key):
|
|
def commands_mock(command, stdin):
|
|
# TODO(maximsmol): this is a hack since stdout=sys.stdout
|
|
# doesn't work with the mock for some reason
|
|
print("ubuntu@ip-.+:~$ exit")
|
|
return PopenBehaviour(stdout="ubuntu@ip-.+:~$ exit")
|
|
|
|
with _setup_popen_mock(commands_mock):
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
scripts.up,
|
|
[
|
|
DEFAULT_TEST_CONFIG_PATH,
|
|
"--no-config-cache",
|
|
"-y",
|
|
"--log-style=pretty",
|
|
"--log-color",
|
|
"False",
|
|
],
|
|
)
|
|
_die_on_error(result)
|
|
|
|
result = runner.invoke(
|
|
scripts.attach,
|
|
[
|
|
DEFAULT_TEST_CONFIG_PATH,
|
|
"--no-config-cache",
|
|
"--log-style=pretty",
|
|
"--log-color",
|
|
"False",
|
|
],
|
|
)
|
|
|
|
_check_output_via_pattern("test_ray_attach.txt", result)
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
sys.platform == "darwin" and "travis" in os.environ.get("USER", ""),
|
|
reason=("Mac builds don't provide proper locale support"),
|
|
)
|
|
def test_ray_attach_with_ip(configure_lang, configure_aws, _unlink_test_ssh_key):
|
|
from ray.autoscaler._private.commands import get_worker_node_ips
|
|
|
|
worker_ip_to_verify = None
|
|
|
|
def commands_mock(command, stdin):
|
|
# TODO(maximsmol): this is a hack since stdout=sys.stdout
|
|
# doesn't work with the mock for some reason
|
|
print("ubuntu@ip-.+:~$ exit")
|
|
return PopenBehaviour(stdout="ubuntu@ip-.+:~$ exit")
|
|
|
|
def commands_verifier(calls):
|
|
for call in calls:
|
|
if len(call[1]) > 0:
|
|
cmd = (
|
|
" ".join(call[1][0]) if isinstance(call[1][0], list) else call[1][0]
|
|
)
|
|
if "ssh" in cmd and worker_ip_to_verify and worker_ip_to_verify in cmd:
|
|
return True
|
|
return False
|
|
|
|
with _setup_popen_mock(commands_mock, commands_verifier):
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
scripts.up,
|
|
[
|
|
DEFAULT_TEST_CONFIG_PATH,
|
|
"--no-config-cache",
|
|
"-y",
|
|
"--log-style=pretty",
|
|
"--log-color",
|
|
"False",
|
|
],
|
|
)
|
|
_die_on_error(result)
|
|
|
|
# Manually create a worker node to test attaching to non-head nodes.
|
|
# ray up only creates the head node; workers are launched asynchronously
|
|
# by the autoscaler which doesn't run in this mocked test environment.
|
|
config = yaml.safe_load(open(DEFAULT_TEST_CONFIG_PATH).read())
|
|
config["cluster_name"] = "test-cli"
|
|
config = prepare_config(config)
|
|
config = aws_config.bootstrap_aws(config)
|
|
|
|
provider = _get_node_provider(config["provider"], config["cluster_name"])
|
|
|
|
worker_node_config = config["available_node_types"]["worker_nodes"][
|
|
"node_config"
|
|
]
|
|
tags = aws_helpers.node_provider_tags(config, "worker_nodes")
|
|
provider.create_node(worker_node_config, tags, 1)
|
|
|
|
worker_ips = get_worker_node_ips(
|
|
DEFAULT_TEST_CONFIG_PATH, override_cluster_name="test-cli"
|
|
)
|
|
assert len(worker_ips) > 0, "Worker node should have been created"
|
|
worker_ip_to_verify = worker_ips[0]
|
|
|
|
result = runner.invoke(
|
|
scripts.attach,
|
|
[
|
|
DEFAULT_TEST_CONFIG_PATH,
|
|
"--no-config-cache",
|
|
"--log-style=pretty",
|
|
"--log-color",
|
|
"False",
|
|
"--node-ip",
|
|
worker_ip_to_verify,
|
|
],
|
|
)
|
|
|
|
_check_output_via_pattern("test_ray_attach_with_ip.txt", result)
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
sys.platform == "darwin" and "travis" in os.environ.get("USER", ""),
|
|
reason=("Mac builds don't provide proper locale support"),
|
|
)
|
|
def test_ray_dashboard(configure_lang, configure_aws, _unlink_test_ssh_key):
|
|
def commands_mock(command, stdin):
|
|
# TODO(maximsmol): this is a hack since stdout=sys.stdout
|
|
# doesn't work with the mock for some reason
|
|
print("ubuntu@ip-.+:~$ exit")
|
|
return PopenBehaviour(stdout="ubuntu@ip-.+:~$ exit")
|
|
|
|
with _setup_popen_mock(commands_mock):
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
scripts.up,
|
|
[
|
|
DEFAULT_TEST_CONFIG_PATH,
|
|
"--no-config-cache",
|
|
"-y",
|
|
"--log-style=pretty",
|
|
"--log-color",
|
|
"False",
|
|
],
|
|
)
|
|
_die_on_error(result)
|
|
|
|
result = runner.invoke(
|
|
scripts.dashboard, [DEFAULT_TEST_CONFIG_PATH, "--no-config-cache"]
|
|
)
|
|
_check_output_via_pattern("test_ray_dashboard.txt", result)
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
sys.platform == "darwin" and "travis" in os.environ.get("USER", ""),
|
|
reason=("Mac builds don't provide proper locale support"),
|
|
)
|
|
def test_ray_exec(configure_lang, configure_aws, _unlink_test_ssh_key):
|
|
def commands_mock(command, stdin):
|
|
# TODO(maximsmol): this is a hack since stdout=sys.stdout
|
|
# doesn't work with the mock for some reason
|
|
print("This is a test!")
|
|
return PopenBehaviour(stdout=b"This is a test!")
|
|
|
|
def commands_verifier(calls):
|
|
for call in calls:
|
|
if len(call[1]) > 0:
|
|
if any(" ray stop; " in token for token in call[1][0]):
|
|
return True
|
|
return False
|
|
|
|
with _setup_popen_mock(commands_mock, commands_verifier):
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
scripts.up,
|
|
[
|
|
DEFAULT_TEST_CONFIG_PATH,
|
|
"--no-config-cache",
|
|
"-y",
|
|
"--log-style=pretty",
|
|
"--log-color",
|
|
"False",
|
|
],
|
|
)
|
|
_die_on_error(result)
|
|
|
|
result = runner.invoke(
|
|
scripts.exec,
|
|
[
|
|
DEFAULT_TEST_CONFIG_PATH,
|
|
"--no-config-cache",
|
|
"--log-style=pretty",
|
|
'"echo This is a test!"',
|
|
"--stop",
|
|
],
|
|
)
|
|
|
|
_check_output_via_pattern("test_ray_exec.txt", result)
|
|
|
|
|
|
# Try to check if we are running in travis. Bazel overrides and controls
|
|
# env vars, so the typical travis env-vars don't help.
|
|
# Unfortunately it will not be nice if your username is travis
|
|
# and you're running on a Mac.
|
|
@pytest.mark.skipif(
|
|
sys.platform == "darwin" and "travis" in os.environ.get("USER", ""),
|
|
reason=("Mac builds don't provide proper locale support"),
|
|
)
|
|
def test_ray_submit(configure_lang, configure_aws, _unlink_test_ssh_key):
|
|
def commands_mock(command, stdin):
|
|
# TODO(maximsmol): this is a hack since stdout=sys.stdout
|
|
# doesn't work with the mock for some reason
|
|
if "rsync" not in command:
|
|
print("This is a test!")
|
|
return PopenBehaviour(stdout=b"This is a test!")
|
|
|
|
with _setup_popen_mock(commands_mock):
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
scripts.up,
|
|
[
|
|
DEFAULT_TEST_CONFIG_PATH,
|
|
"--no-config-cache",
|
|
"-y",
|
|
"--log-style=pretty",
|
|
"--log-color",
|
|
"False",
|
|
],
|
|
)
|
|
_die_on_error(result)
|
|
|
|
with tempfile.NamedTemporaryFile(suffix="test.py", mode="w") as f:
|
|
f.write("print('This is a test!')\n")
|
|
result = runner.invoke(
|
|
scripts.submit,
|
|
[
|
|
DEFAULT_TEST_CONFIG_PATH,
|
|
"--no-config-cache",
|
|
"--log-style=pretty",
|
|
"--log-color",
|
|
"False",
|
|
# this is somewhat misleading, since the file
|
|
# actually never gets run
|
|
# TODO(maximsmol): make this work properly one day?
|
|
f.name,
|
|
],
|
|
)
|
|
|
|
_check_output_via_pattern("test_ray_submit.txt", result)
|
|
|
|
|
|
@pytest.mark.parametrize("enable_v2", [True, False])
|
|
def test_ray_status(shutdown_only, monkeypatch, enable_v2):
|
|
address = ray.init(
|
|
num_cpus=3, _system_config={"enable_autoscaler_v2": enable_v2}
|
|
).get("address")
|
|
runner = CliRunner()
|
|
|
|
def output_ready():
|
|
result = runner.invoke(scripts.status)
|
|
_ = result.stdout
|
|
if not result.exception and "memory" in result.output:
|
|
return True
|
|
raise RuntimeError(
|
|
f"result.exception={result.exception} " f"result.output={result.output}"
|
|
)
|
|
|
|
wait_for_condition(output_ready)
|
|
|
|
# Wait one whole reporting cycle
|
|
time.sleep(autoscaler_constants.AUTOSCALER_UPDATE_INTERVAL_S)
|
|
if enable_v2:
|
|
filename_expected = "test_ray_status.txt"
|
|
else:
|
|
filename_expected = "test_ray_status_v1.txt"
|
|
|
|
result = runner.invoke(scripts.status, [])
|
|
_check_output_via_pattern(filename_expected, result)
|
|
|
|
result_arg = runner.invoke(scripts.status, ["--address", address])
|
|
|
|
_check_output_via_pattern(filename_expected, result_arg)
|
|
|
|
# Try to check status with RAY_ADDRESS set
|
|
monkeypatch.setenv("RAY_ADDRESS", address)
|
|
result_env = runner.invoke(scripts.status)
|
|
_check_output_via_pattern(filename_expected, result_env)
|
|
|
|
result_env_arg = runner.invoke(scripts.status, ["--address", address])
|
|
_check_output_via_pattern(filename_expected, result_env_arg)
|
|
|
|
|
|
@pytest.mark.xfail(cluster_not_supported, reason="cluster not supported on Windows")
|
|
@pytest.mark.parametrize("enable_v2", [True, False])
|
|
def test_ray_status_multinode(ray_start_cluster, enable_v2):
|
|
cluster = ray_start_cluster
|
|
cluster.add_node(num_cpus=2, _system_config={"enable_autoscaler_v2": enable_v2})
|
|
ray.init(address=cluster.address)
|
|
for _ in range(3):
|
|
cluster.add_node(num_cpus=2)
|
|
wait_for_condition(lambda: len(list_nodes()) == 4)
|
|
runner = CliRunner()
|
|
|
|
# Wait one whole reporting cycle
|
|
time.sleep(autoscaler_constants.AUTOSCALER_UPDATE_INTERVAL_S)
|
|
|
|
def output_ready():
|
|
result = runner.invoke(scripts.status)
|
|
_ = result.stdout
|
|
if not result.exception and "memory" in result.output:
|
|
return True
|
|
raise RuntimeError(
|
|
f"result.exception={result.exception} " f"result.output={result.output}"
|
|
)
|
|
|
|
wait_for_condition(output_ready)
|
|
|
|
result = runner.invoke(scripts.status, [])
|
|
if enable_v2:
|
|
_check_output_via_pattern("test_ray_status_multinode.txt", result)
|
|
else:
|
|
_check_output_via_pattern("test_ray_status_multinode_v1.txt", result)
|
|
|
|
|
|
@pytest.fixture
|
|
def start_open_port_check_server():
|
|
class OpenPortCheckServer(BaseHTTPRequestHandler):
|
|
request_ports = None
|
|
response_open_ports = []
|
|
|
|
def do_POST(self):
|
|
content_length = int(self.headers["Content-Length"])
|
|
post_data = self.rfile.read(content_length)
|
|
payload = json.loads(post_data)
|
|
OpenPortCheckServer.request_ports = payload["ports"]
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(
|
|
json.dumps(
|
|
{
|
|
"open_ports": OpenPortCheckServer.response_open_ports,
|
|
"checked_ports": payload["ports"],
|
|
}
|
|
).encode("utf-8")
|
|
)
|
|
|
|
server = HTTPServer(("127.0.0.1", 0), OpenPortCheckServer)
|
|
server_thread = threading.Thread(target=server.serve_forever)
|
|
server_thread.start()
|
|
|
|
yield (
|
|
OpenPortCheckServer,
|
|
f"http://{build_address(server.server_address[0], server.server_address[1])}",
|
|
)
|
|
|
|
server.shutdown()
|
|
server_thread.join()
|
|
|
|
|
|
def test_ray_check_open_ports(shutdown_only, start_open_port_check_server):
|
|
context = ray.init()
|
|
|
|
open_port_check_server, url = start_open_port_check_server
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
check_open_ports,
|
|
[
|
|
"-y",
|
|
"--service-url",
|
|
url,
|
|
],
|
|
)
|
|
assert result.exit_code == 0
|
|
assert (
|
|
int(parse_address(context.address_info["gcs_address"])[1])
|
|
in open_port_check_server.request_ports
|
|
)
|
|
assert "[🟢] No open ports detected" in result.output
|
|
|
|
open_port_check_server.response_open_ports = [
|
|
context.address_info["metrics_export_port"]
|
|
]
|
|
result = runner.invoke(
|
|
check_open_ports,
|
|
[
|
|
"-y",
|
|
"--service-url",
|
|
url,
|
|
],
|
|
)
|
|
assert result.exit_code == 0
|
|
assert "[🛑] open ports detected" in result.output
|
|
|
|
|
|
def test_ray_drain_node(monkeypatch, shutdown_only):
|
|
monkeypatch.setenv("RAY_py_gcs_connect_timeout_s", "1")
|
|
ray.init()
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
scripts.drain_node,
|
|
[
|
|
"--node-id",
|
|
"0db0438b5cfd6e84d7ec07212ed76b23be7886cbd426ef4d1879bebf",
|
|
"--reason",
|
|
"DRAIN_NODE_REASON_IDLE_TERMINATION",
|
|
"--reason-message",
|
|
"idle termination",
|
|
"--deadline-remaining-seconds",
|
|
"-1",
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "--deadline-remaining-seconds cannot be negative, got -1" in result.output
|
|
|
|
# Test invalid drain reason.
|
|
result = runner.invoke(
|
|
scripts.drain_node,
|
|
[
|
|
"--node-id",
|
|
"0db0438b5cfd6e84d7ec07212ed76b23be7886cbd426ef4d1879bebf",
|
|
"--reason",
|
|
"DRAIN_NODE_REASON_INVALID",
|
|
"--reason-message",
|
|
"idle termination",
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert (
|
|
"is not one of 'DRAIN_NODE_REASON_IDLE_TERMINATION', "
|
|
"'DRAIN_NODE_REASON_PREEMPTION'" in result.output
|
|
)
|
|
|
|
result = runner.invoke(
|
|
scripts.drain_node,
|
|
[
|
|
"--address",
|
|
"127.0.0.2:8888",
|
|
"--node-id",
|
|
"0db0438b5cfd6e84d7ec07212ed76b23be7886cbd426ef4d1879bebf",
|
|
"--reason",
|
|
"DRAIN_NODE_REASON_IDLE_TERMINATION",
|
|
"--reason-message",
|
|
"idle termination",
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "Timed out while waiting for GCS to become available" in str(
|
|
result.exception
|
|
)
|
|
|
|
result = runner.invoke(
|
|
scripts.drain_node,
|
|
[
|
|
"--node-id",
|
|
"invalid-node-id",
|
|
"--reason",
|
|
"DRAIN_NODE_REASON_IDLE_TERMINATION",
|
|
"--reason-message",
|
|
"idle termination",
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "Invalid hex ID of a Ray node, got invalid-node-id" in result.output
|
|
|
|
with patch("ray._raylet.GcsClient") as MockGcsClient:
|
|
mock_gcs_client = MockGcsClient.return_value
|
|
mock_gcs_client.internal_kv_get.return_value = (
|
|
'{"ray_version": "ray_version_mismatch"}'.encode()
|
|
)
|
|
result = runner.invoke(
|
|
scripts.drain_node,
|
|
[
|
|
"--address",
|
|
"127.0.0.1:6543",
|
|
"--node-id",
|
|
"0db0438b5cfd6e84d7ec07212ed76b23be7886cbd426ef4d1879bebf",
|
|
"--reason",
|
|
"DRAIN_NODE_REASON_IDLE_TERMINATION",
|
|
"--reason-message",
|
|
"idle termination",
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "Ray version mismatch" in str(result.exception)
|
|
|
|
with patch("ray._raylet.GcsClient") as MockGcsClient:
|
|
mock_gcs_client = MockGcsClient.return_value
|
|
mock_gcs_client.internal_kv_get.return_value = (
|
|
f'{{"ray_version": "{ray.__version__}"}}'.encode()
|
|
)
|
|
mock_gcs_client.drain_node.return_value = (True, "")
|
|
result = runner.invoke(
|
|
scripts.drain_node,
|
|
[
|
|
"--address",
|
|
"127.0.0.1:6543",
|
|
"--node-id",
|
|
"0db0438b5cfd6e84d7ec07212ed76b23be7886cbd426ef4d1879bebf",
|
|
"--reason",
|
|
"DRAIN_NODE_REASON_IDLE_TERMINATION",
|
|
"--reason-message",
|
|
"idle termination",
|
|
],
|
|
)
|
|
assert result.exit_code == 0
|
|
assert mock_gcs_client.mock_calls[1] == mock.call.drain_node(
|
|
"0db0438b5cfd6e84d7ec07212ed76b23be7886cbd426ef4d1879bebf",
|
|
1,
|
|
"idle termination",
|
|
0,
|
|
)
|
|
|
|
with patch("ray._raylet.GcsClient") as MockGcsClient:
|
|
mock_gcs_client = MockGcsClient.return_value
|
|
mock_gcs_client.internal_kv_get.return_value = (
|
|
f'{{"ray_version": "{ray.__version__}"}}'.encode()
|
|
)
|
|
mock_gcs_client.drain_node.return_value = (False, "Node not idle")
|
|
result = runner.invoke(
|
|
scripts.drain_node,
|
|
[
|
|
"--address",
|
|
"127.0.0.1:6543",
|
|
"--node-id",
|
|
"0db0438b5cfd6e84d7ec07212ed76b23be7886cbd426ef4d1879bebf",
|
|
"--reason",
|
|
"DRAIN_NODE_REASON_IDLE_TERMINATION",
|
|
"--reason-message",
|
|
"idle termination",
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "The drain request is not accepted: Node not idle" in result.output
|
|
|
|
# Test without node-id
|
|
with patch("ray._raylet.GcsClient") as MockGcsClient:
|
|
mock_gcs_client = MockGcsClient.return_value
|
|
mock_gcs_client.internal_kv_get.return_value = (
|
|
f'{{"ray_version": "{ray.__version__}"}}'.encode()
|
|
)
|
|
mock_gcs_client.drain_node.return_value = (True, "")
|
|
result = runner.invoke(
|
|
scripts.drain_node,
|
|
[
|
|
"--address",
|
|
"127.0.01:6543",
|
|
"--reason",
|
|
"DRAIN_NODE_REASON_PREEMPTION",
|
|
"--reason-message",
|
|
"spot preemption",
|
|
],
|
|
)
|
|
assert result.exit_code == 0
|
|
assert mock_gcs_client.mock_calls[1] == mock.call.drain_node(
|
|
ray.get_runtime_context().get_node_id(), 2, "spot preemption", 0
|
|
)
|
|
|
|
with patch("time.time_ns", return_value=1000000000), patch(
|
|
"ray._raylet.GcsClient"
|
|
) as MockGcsClient:
|
|
mock_gcs_client = MockGcsClient.return_value
|
|
mock_gcs_client.internal_kv_get.return_value = (
|
|
f'{{"ray_version": "{ray.__version__}"}}'.encode()
|
|
)
|
|
mock_gcs_client.drain_node.return_value = (True, "")
|
|
result = runner.invoke(
|
|
scripts.drain_node,
|
|
[
|
|
"--address",
|
|
"127.0.0.1:6543",
|
|
"--node-id",
|
|
"0db0438b5cfd6e84d7ec07212ed76b23be7886cbd426ef4d1879bebf",
|
|
"--reason",
|
|
"DRAIN_NODE_REASON_PREEMPTION",
|
|
"--reason-message",
|
|
"spot preemption",
|
|
"--deadline-remaining-seconds",
|
|
"1",
|
|
],
|
|
)
|
|
assert result.exit_code == 0
|
|
assert mock_gcs_client.mock_calls[1] == mock.call.drain_node(
|
|
"0db0438b5cfd6e84d7ec07212ed76b23be7886cbd426ef4d1879bebf",
|
|
2,
|
|
"spot preemption",
|
|
2000,
|
|
)
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
sys.platform == "darwin" and "travis" in os.environ.get("USER", ""),
|
|
reason=("Mac builds don't provide proper locale support"),
|
|
)
|
|
def test_ray_cluster_dump(configure_lang, configure_aws, _unlink_test_ssh_key):
|
|
def commands_mock(command, stdin):
|
|
print("This is a test!")
|
|
return PopenBehaviour(stdout=b"This is a test!")
|
|
|
|
with _setup_popen_mock(commands_mock):
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
scripts.up,
|
|
[
|
|
DEFAULT_TEST_CONFIG_PATH,
|
|
"--no-config-cache",
|
|
"-y",
|
|
"--log-style=pretty",
|
|
"--log-color",
|
|
"False",
|
|
],
|
|
)
|
|
_die_on_error(result)
|
|
|
|
result = runner.invoke(
|
|
scripts.cluster_dump, [DEFAULT_TEST_CONFIG_PATH, "--no-processes"]
|
|
)
|
|
|
|
_check_output_via_pattern("test_ray_cluster_dump.txt", result)
|
|
|
|
|
|
def test_kill_actor_by_name_via_cli(ray_start_regular):
|
|
"""Test killing a named actor via CLI (both force and graceful termination). Covers regular and detached actors."""
|
|
address = ray_start_regular["address"]
|
|
runner = CliRunner()
|
|
# Kill non-existing actor
|
|
result = runner.invoke(
|
|
scripts.kill_actor,
|
|
[
|
|
"--address",
|
|
address,
|
|
"--name",
|
|
"non_existing_actor",
|
|
"--namespace",
|
|
"ns",
|
|
"--force",
|
|
],
|
|
)
|
|
assert result.exit_code == 1
|
|
assert "No named actor found: non_existing_actor (namespace=ns)" in result.output
|
|
|
|
@ray.remote(max_restarts=-1)
|
|
class Actor:
|
|
def ping(self):
|
|
return "pong"
|
|
|
|
def crash(self):
|
|
raise RuntimeError("intentional")
|
|
|
|
# Regular named actor with namespace
|
|
# Kill without --no-restart flag
|
|
actor = Actor.options(name="test_actor", namespace="ns").remote()
|
|
assert ray.util.list_named_actors(all_namespaces=True) == [
|
|
{"name": "test_actor", "namespace": "ns"}
|
|
]
|
|
try:
|
|
ray.get(actor.crash.remote())
|
|
except ray.exceptions.RayTaskError:
|
|
pass
|
|
wait_for_condition(lambda: ray.get(actor.ping.remote()) == "pong", timeout=10)
|
|
|
|
result = runner.invoke(
|
|
scripts.kill_actor,
|
|
[
|
|
"--address",
|
|
address,
|
|
"--name",
|
|
"test_actor",
|
|
"--namespace",
|
|
"ns",
|
|
"--force",
|
|
],
|
|
)
|
|
_die_on_error(result)
|
|
wait_for_condition(
|
|
lambda: ray.util.list_named_actors(all_namespaces=True)
|
|
== [{"name": "test_actor", "namespace": "ns"}],
|
|
timeout=10,
|
|
)
|
|
wait_for_condition(lambda: ray.get(actor.ping.remote()) == "pong", timeout=10)
|
|
# Kill with --no-restart flag
|
|
result = runner.invoke(
|
|
scripts.kill_actor,
|
|
[
|
|
"--address",
|
|
address,
|
|
"--name",
|
|
"test_actor",
|
|
"--namespace",
|
|
"ns",
|
|
"--force",
|
|
"--no-restart",
|
|
],
|
|
)
|
|
_die_on_error(result)
|
|
wait_for_condition(
|
|
lambda: ray.util.list_named_actors(all_namespaces=True) == [], timeout=10
|
|
)
|
|
|
|
# Detached named actor
|
|
detached_actor = Actor.options(
|
|
name="detached_test_actor", namespace="ns", lifetime="detached"
|
|
).remote()
|
|
assert ray.get(detached_actor.ping.remote()) == "pong"
|
|
assert ray.util.list_named_actors(all_namespaces=True) == [
|
|
{"name": "detached_test_actor", "namespace": "ns"}
|
|
]
|
|
result = runner.invoke(
|
|
scripts.kill_actor,
|
|
[
|
|
"--address",
|
|
address,
|
|
"--name",
|
|
"detached_test_actor",
|
|
"--namespace",
|
|
"ns",
|
|
"--force",
|
|
"--no-restart",
|
|
],
|
|
)
|
|
_die_on_error(result)
|
|
wait_for_condition(
|
|
lambda: ray.util.list_named_actors(all_namespaces=True) == [], timeout=10
|
|
)
|
|
# Test graceful termination of named actor
|
|
def check_killed(actorhandle):
|
|
try:
|
|
ray.get(actorhandle.ping.remote(), timeout=1)
|
|
return False
|
|
except Exception:
|
|
return True
|
|
|
|
graceful_actor = Actor.options(name="graceful_actor", namespace="ns").remote()
|
|
assert ray.get(graceful_actor.ping.remote()) == "pong"
|
|
|
|
result = runner.invoke(
|
|
scripts.kill_actor,
|
|
[
|
|
"--address",
|
|
address,
|
|
"--name",
|
|
"graceful_actor",
|
|
"--namespace",
|
|
"ns",
|
|
],
|
|
)
|
|
_die_on_error(result)
|
|
wait_for_condition(lambda: check_killed(graceful_actor), timeout=10)
|
|
wait_for_condition(
|
|
lambda: ray.util.list_named_actors(all_namespaces=True) == [], timeout=10
|
|
)
|
|
|
|
|
|
# Check if ray.serve is available (it's an optional dependency)
|
|
try:
|
|
import ray.serve.scripts # noqa: F401
|
|
|
|
SERVE_AVAILABLE = True
|
|
except ImportError:
|
|
SERVE_AVAILABLE = False
|
|
|
|
|
|
@pytest.mark.skipif(not SERVE_AVAILABLE, reason="ray.serve is not installed")
|
|
def test_serve_cli_registered():
|
|
"""Test that serve CLI is properly registered with the main ray CLI."""
|
|
# Verify 'serve' command is registered
|
|
assert "serve" in scripts.cli.commands, "serve command should be registered"
|
|
|
|
# Verify serve subcommands are accessible
|
|
serve_cli = scripts.cli.commands["serve"]
|
|
expected_commands = {
|
|
"start",
|
|
"deploy",
|
|
"run",
|
|
"config",
|
|
"status",
|
|
"shutdown",
|
|
"build",
|
|
}
|
|
assert expected_commands.issubset(set(serve_cli.commands.keys()))
|
|
|
|
|
|
@pytest.mark.skipif(not SERVE_AVAILABLE, reason="ray.serve is not installed")
|
|
def test_ray_serve_help():
|
|
"""Test that 'ray serve --help' works correctly."""
|
|
runner = CliRunner()
|
|
result = runner.invoke(scripts.cli, ["serve", "--help"])
|
|
assert result.exit_code == 0, result.output
|
|
assert "CLI for managing Serve applications" in result.output
|
|
assert "deploy" in result.output
|
|
assert "run" in result.output
|
|
|
|
|
|
@pytest.mark.skipif(sys.platform != "win32", reason="Windows-only agent cleanup test")
|
|
def test_ray_stop_terminates_agents_on_windows(cleanup_ray):
|
|
"""Verify that `ray stop` terminates dashboard and runtime_env agent
|
|
processes on Windows, where setproctitle does not change the process
|
|
name or command line visible to psutil.
|
|
|
|
See https://github.com/ray-project/ray/issues/61452
|
|
"""
|
|
runner = CliRunner(env={"RAY_USAGE_STATS_PROMPT_ENABLED": "0"})
|
|
|
|
# Start a Ray head node with the dashboard enabled.
|
|
result = runner.invoke(
|
|
scripts.start,
|
|
["--head", "--include-dashboard=true"],
|
|
)
|
|
_die_on_error(result)
|
|
|
|
agent_keywords = [
|
|
os.path.join("dashboard", "agent.py"),
|
|
os.path.join("runtime_env", "agent", "main.py"),
|
|
]
|
|
|
|
def _find_agent_procs():
|
|
"""Return agent processes whose cmdline contains one of the keywords."""
|
|
agents = []
|
|
for proc in psutil.process_iter(["pid", "cmdline"]):
|
|
try:
|
|
cmdline_str = " ".join(proc.cmdline())
|
|
except (psutil.NoSuchProcess, psutil.AccessDenied):
|
|
continue
|
|
if any(kw in cmdline_str for kw in agent_keywords):
|
|
agents.append(proc)
|
|
return agents
|
|
|
|
# Wait for both agents to be running.
|
|
wait_for_condition(lambda: len(_find_agent_procs()) >= 2, timeout=30)
|
|
|
|
# Stop Ray.
|
|
stop_result = runner.invoke(scripts.stop)
|
|
_die_on_error(stop_result)
|
|
|
|
# Verify that all agent processes have exited.
|
|
def _agents_gone():
|
|
remaining = _find_agent_procs()
|
|
return len(remaining) == 0
|
|
|
|
wait_for_condition(_agents_gone, timeout=30)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(pytest.main(["-sv", __file__]))
|