wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-18 12:40:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

[autoscaler/docker] Migrate --docker to --run-env (#9001)

Browse Source
This commit is contained in:
Ian Rodney
2020-06-25 20:37:42 -07:00
committed by GitHub
parent 11bda7272b
commit 42631c5b87
5 changed files with 94 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/ray/autoscaler/commands.py
+35 -33
View File
@@ -23,16 +23,18 @@ from ray.autoscaler.util import validate_config, hash_runtime_conf, \
from ray.autoscaler.node_provider import get_node_provider, NODE_PROVIDERS
from ray.autoscaler.tags import TAG_RAY_NODE_TYPE, TAG_RAY_LAUNCH_CONFIG, \
TAG_RAY_NODE_NAME, NODE_TYPE_WORKER, NODE_TYPE_HEAD
from ray.ray_constants import AUTOSCALER_RESOURCE_REQUEST_CHANNEL
from ray.autoscaler.updater import NodeUpdaterThread
from ray.autoscaler.updater import NodeUpdaterThread, DockerCommandRunner
from ray.autoscaler.log_timer import LogTimer
from ray.autoscaler.docker import with_docker_exec
from ray.worker import global_worker
logger = logging.getLogger(__name__)
redis_client = None
RUN_ENV_TYPES = ["auto", "host", "docker"]
def _redis():
global redis_client
@@ -226,9 +228,9 @@ def kill_node(config_file, yes, hard, override_cluster_name):
def monitor_cluster(cluster_config_file, num_lines, override_cluster_name):
"""Kills a random Raylet worker."""
"""Tails the autoscaler logs of a Ray cluster."""
cmd = "tail -n {} -f /tmp/ray/session_*/logs/monitor*".format(num_lines)
exec_cluster(cluster_config_file, cmd, False, False, False, False, False,
exec_cluster(cluster_config_file, cmd, "auto", False, False, False, False,
override_cluster_name, None)
@@ -369,17 +371,14 @@ def get_or_create_head_node(config, config_file, no_restart, restart_only, yes,
"Head node up-to-date, IP address is: {}".format(head_node_ip))
monitor_str = "tail -n 100 -f /tmp/ray/session_*/logs/monitor*"
use_docker = "docker" in config and bool(
config["docker"]["container_name"])
if override_cluster_name:
modifiers = " --cluster-name={}".format(
quote(override_cluster_name))
else:
modifiers = ""
print("To monitor auto-scaling activity, you can run:\n\n"
" ray exec {} {}{}{}\n".format(
config_file, "--docker " if use_docker else "",
quote(monitor_str), modifiers))
" ray exec {} {}{}\n".format(config_file, quote(monitor_str),
modifiers))
print("To open a console on the cluster:\n\n"
" ray attach {}{}\n".format(config_file, modifiers))
@@ -419,13 +418,13 @@ def attach_cluster(config_file, start, use_screen, use_tmux,
"--new only makes sense if passing --screen or --tmux")
cmd = "$SHELL"
exec_cluster(config_file, cmd, False, False, False, False, start,
exec_cluster(config_file, cmd, "auto", False, False, False, start,
override_cluster_name, port_forward)
def exec_cluster(config_file,
cmd=None,
docker=False,
run_env="auto",
screen=False,
tmux=False,
stop=False,
@@ -438,7 +437,8 @@ def exec_cluster(config_file,
Arguments:
config_file: path to the cluster yaml
cmd: command to run
docker: whether to run command in docker container of config
run_env: whether to run the command on the host or in a container.
Select between "auto", "host" and "docker"
screen: whether to run in a screen
tmux: whether to run in a tmux session
stop: whether to stop the cluster after command run
@@ -447,7 +447,8 @@ def exec_cluster(config_file,
port_forward (int or list[int]): port(s) to forward
"""
assert not (screen and tmux), "Can specify only one of `screen` or `tmux`."
assert run_env in RUN_ENV_TYPES, "--run_env must be in {}".format(
RUN_ENV_TYPES)
config = yaml.safe_load(open(config_file).read())
if override_cluster_name is not None:
config["cluster_name"] = override_cluster_name
@@ -471,23 +472,17 @@ def exec_cluster(config_file,
runtime_hash="",
docker_config=config.get("docker"))
def wrap_docker(command):
container_name = config["docker"]["container_name"]
if not container_name:
raise ValueError("Docker container not specified in config.")
return with_docker_exec(
[command], container_name=container_name)[0]
is_docker = isinstance(updater.cmd_runner, DockerCommandRunner)
if cmd:
cmd = wrap_docker(cmd) if docker else cmd
if stop:
shutdown_cmd = (
"ray stop; ray teardown ~/ray_bootstrap_config.yaml "
"--yes --workers-only")
if docker:
shutdown_cmd = wrap_docker(shutdown_cmd)
cmd += ("; {}; sudo shutdown -h now".format(shutdown_cmd))
if cmd and stop:
cmd += "; ".join([
"ray stop",
"ray teardown ~/ray_bootstrap_config.yaml --yes --workers-only"
])
if is_docker and run_env == "docker":
updater.cmd_runner.shutdown_after_next_cmd()
else:
cmd += "; sudo shutdown -h now"
result = _exec(
updater,
@@ -495,8 +490,8 @@ def exec_cluster(config_file,
screen,
tmux,
port_forward=port_forward,
with_output=with_output)
with_output=with_output,
run_env=run_env)
if tmux or screen:
attach_command_parts = ["ray attach", config_file]
if override_cluster_name is not None:
@@ -516,7 +511,13 @@ def exec_cluster(config_file,
provider.cleanup()
def _exec(updater, cmd, screen, tmux, port_forward=None, with_output=False):
def _exec(updater,
cmd,
screen,
tmux,
port_forward=None,
with_output=False,
run_env="auto"):
if cmd:
if screen:
cmd = [
@@ -535,7 +536,8 @@ def _exec(updater, cmd, screen, tmux, port_forward=None, with_output=False):
cmd,
exit_on_fail=True,
port_forward=port_forward,
with_output=with_output)
with_output=with_output,
run_env=run_env)
def rsync(config_file,
python/ray/autoscaler/docker.py
+11 -10
View File
@@ -62,24 +62,24 @@ def dockerize_if_needed(config):
return config
def with_docker_exec(cmds, container_name, env_vars=None):
def with_docker_exec(cmds,
container_name,
env_vars=None,
with_interactive=False):
env_str = ""
if env_vars:
env_str = " ".join(
["-e {env}=${env}".format(env=env) for env in env_vars])
return [
"docker exec {} {} /bin/sh -c {} ".format(env_str, container_name,
quote(cmd)) for cmd in cmds
"docker exec {interactive} {env} {container} /bin/bash -c {cmd} ".
format(
interactive="-it" if with_interactive else "",
env=env_str,
container=container_name,
cmd=quote(cmd)) for cmd in cmds
]
def aptwait_cmd():
return ("while sudo fuser"
" /var/{lib/{dpkg,apt/lists},cache/apt/archives}/lock"
" >/dev/null 2>&1; "
"do echo 'Waiting for release of dpkg/apt locks'; sleep 5; done")
def check_docker_running_cmd(cname):
return " ".join(["docker", "inspect", "-f", "'{{.State.Running}}'", cname])
@@ -102,6 +102,7 @@ def docker_start_cmds(user, image, mount, cname, user_options):
["-e {name}={val}".format(name=k, val=v) for k, v in env_vars.items()])
user_options_str = " ".join(user_options)
# TODO(ilr) Check command type
# docker run command
docker_check = check_docker_running_cmd(cname) + " || "
docker_run = [
python/ray/autoscaler/updater.py
+34 -10
View File
@@ -17,7 +17,7 @@ from ray.autoscaler.tags import TAG_RAY_NODE_STATUS, TAG_RAY_RUNTIME_CONFIG, \
STATUS_UP_TO_DATE, STATUS_UPDATE_FAILED, STATUS_WAITING_FOR_SSH, \
STATUS_SETTING_UP, STATUS_SYNCING_FILES
from ray.autoscaler.log_timer import LogTimer
from ray.autoscaler.docker import check_docker_running_cmd
from ray.autoscaler.docker import check_docker_running_cmd, with_docker_exec
logger = logging.getLogger(__name__)
@@ -50,7 +50,8 @@ class KubernetesCommandRunner:
timeout=120,
exit_on_fail=False,
port_forward=None,
with_output=False):
with_output=False,
**kwargs):
if cmd and port_forward:
raise Exception(
"exec with Kubernetes can't forward ports and execute"
@@ -239,7 +240,8 @@ class SSHCommandRunner:
timeout=120,
exit_on_fail=False,
port_forward=None,
with_output=False):
with_output=False,
**kwargs):
self.set_ssh_ip_if_required()
@@ -307,14 +309,27 @@ class DockerCommandRunner(SSHCommandRunner):
self.docker_name = docker_config["container_name"]
self.docker_config = docker_config
self.home_dir = None
self.shutdown = False
def run(self,
cmd,
timeout=120,
exit_on_fail=False,
port_forward=None,
with_output=False):
with_output=False,
run_env=True,
**kwargs):
if run_env == "auto":
run_env = "host" if cmd.find("docker") == 0 else "docker"
if run_env == "docker":
cmd = self.docker_expand_user(cmd, any_char=True)
cmd = with_docker_exec(
[cmd], container_name=self.docker_name,
with_interactive=True)[0]
if self.shutdown:
cmd += "; sudo shutdown -h now"
return self.ssh_command_runner.run(
cmd,
timeout=timeout,
@@ -322,6 +337,9 @@ class DockerCommandRunner(SSHCommandRunner):
port_forward=None,
with_output=False)
def shutdown_after_next_cmd(self):
self.shutdown = True
def check_container_status(self):
no_exist = "not_present"
cmd = check_docker_running_cmd(self.docker_name) + " ".join(
@@ -330,7 +348,7 @@ class DockerCommandRunner(SSHCommandRunner):
cmd, with_output=True).decode("utf-8").strip()
if no_exist in output:
return False
return output
return "true" in output.lower()
def run_rsync_up(self, source, target):
self.ssh_command_runner.run_rsync_up(source, target)
@@ -349,16 +367,22 @@ class DockerCommandRunner(SSHCommandRunner):
return inner_str + " docker exec -it {} /bin/bash\n".format(
self.docker_name)
def docker_expand_user(self, string):
if string.find("~") == 0:
def docker_expand_user(self, string, any_char=False):
user_pos = string.find("~")
if user_pos > -1:
if self.home_dir is None:
self.home_dir = self.ssh_command_runner.run(
"docker exec {} env | grep HOME | cut -d'=' -f2".format(
self.docker_name),
with_output=True).decode("utf-8").strip()
return string.replace("~", self.home_dir)
else:
return string
if any_char:
return string.replace("~/", self.home_dir + "/")
elif not any_char and user_pos == 0:
return string.replace("~", self.home_dir, 1)
return string
class NodeUpdater:
python/ray/projects/scripts.py
+1 -1
View File
@@ -229,7 +229,7 @@ class SessionRunner:
exec_cluster(
config_file=self.project_definition.cluster_yaml(),
cmd=cmd,
docker=False,
run_env=config.get("run_env", "auto"),
screen=False,
tmux=config.get("tmux", False),
stop=False,
python/ray/scripts/scripts.py
+13 -17
View File
@@ -16,7 +16,7 @@ import ray.services as services
from ray.autoscaler.commands import (
attach_cluster, exec_cluster, create_or_update_cluster, monitor_cluster,
rsync, teardown_cluster, get_head_node_ip, kill_node, get_worker_node_ips,
debug_status)
debug_status, RUN_ENV_TYPES)
import ray.ray_constants as ray_constants
import ray.utils
from ray.projects.scripts import project_cli, session_cli
@@ -831,11 +831,6 @@ def rsync_up(cluster_config_file, source, target, cluster_name, all_nodes):
@cli.command(context_settings={"ignore_unknown_options": True})
@click.argument("cluster_config_file", required=True, type=str)
@click.option(
"--docker",
is_flag=True,
default=False,
help="Runs command in the docker container specified in cluster_config.")
@click.option(
"--stop",
is_flag=True,
@@ -873,8 +868,8 @@ def rsync_up(cluster_config_file, source, target, cluster_name, all_nodes):
type=str,
help="(deprecated) Use '-- --arg1 --arg2' for script args.")
@click.argument("script_args", nargs=-1)
def submit(cluster_config_file, docker, screen, tmux, stop, start,
cluster_name, port_forward, script, args, script_args):
def submit(cluster_config_file, screen, tmux, stop, start, cluster_name,
port_forward, script, args, script_args):
"""Uploads and runs a script on the specified cluster.
The script is automatically synced to the following location:
@@ -897,8 +892,7 @@ def submit(cluster_config_file, docker, screen, tmux, stop, start,
create_or_update_cluster(cluster_config_file, None, None, False, False,
True, cluster_name)
target = os.path.basename(script)
if not docker:
target = os.path.join("~", target)
target = os.path.join("~", target)
rsync(cluster_config_file, script, target, cluster_name, down=False)
command_parts = ["python", target]
@@ -912,7 +906,7 @@ def submit(cluster_config_file, docker, screen, tmux, stop, start,
exec_cluster(
cluster_config_file,
cmd,
docker,
"docker",
screen,
tmux,
stop,
@@ -925,10 +919,12 @@ def submit(cluster_config_file, docker, screen, tmux, stop, start,
@click.argument("cluster_config_file", required=True, type=str)
@click.argument("cmd", required=True, type=str)
@click.option(
"--docker",
is_flag=True,
default=False,
help="Runs command in the docker container specified in cluster_config.")
"--run-env",
required=False,
type=click.Choice(RUN_ENV_TYPES),
default="auto",
help="Choose whether to execute this command in a container or directly on"
" the cluster head. Only applies when docker is configured in the YAML.")
@click.option(
"--stop",
is_flag=True,
@@ -959,11 +955,11 @@ def submit(cluster_config_file, docker, screen, tmux, stop, start,
multiple=True,
type=int,
help="Port to forward. Use this multiple times to forward multiple ports.")
def exec_cmd(cluster_config_file, cmd, docker, screen, tmux, stop, start,
def exec_cmd(cluster_config_file, cmd, run_env, screen, tmux, stop, start,
cluster_name, port_forward):
"""Execute a command via SSH on a Ray cluster."""
port_forward = [(port, port) for port in list(port_forward)]
exec_cluster(cluster_config_file, cmd, docker, screen, tmux, stop, start,
exec_cluster(cluster_config_file, cmd, run_env, screen, tmux, stop, start,
cluster_name, port_forward)