diff --git a/python/ray/autoscaler/commands.py b/python/ray/autoscaler/commands.py index a419977cc..0c70c1368 100644 --- a/python/ray/autoscaler/commands.py +++ b/python/ray/autoscaler/commands.py @@ -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, diff --git a/python/ray/autoscaler/docker.py b/python/ray/autoscaler/docker.py index 8973c253f..ac88778e4 100644 --- a/python/ray/autoscaler/docker.py +++ b/python/ray/autoscaler/docker.py @@ -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 = [ diff --git a/python/ray/autoscaler/updater.py b/python/ray/autoscaler/updater.py index 589d4388e..049ab3e45 100644 --- a/python/ray/autoscaler/updater.py +++ b/python/ray/autoscaler/updater.py @@ -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: diff --git a/python/ray/projects/scripts.py b/python/ray/projects/scripts.py index 2610f3232..faff5ee28 100644 --- a/python/ray/projects/scripts.py +++ b/python/ray/projects/scripts.py @@ -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, diff --git a/python/ray/scripts/scripts.py b/python/ray/scripts/scripts.py index 99ce78807..276396de9 100644 --- a/python/ray/scripts/scripts.py +++ b/python/ray/scripts/scripts.py @@ -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)