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

[docker] Fix restart behavior with Docker (#12898)

Browse Source 
Co-authored-by: Richard Liaw <rliaw@berkeley.edu>
Co-authored-by: ijrsvt <ilr@anyscale.com>
This commit is contained in:
Ian Rodney
2020-12-28 18:56:28 -08:00
committed by GitHub
parent d1dd3410c8
commit 7ad56826db
4 changed files with 327 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/ray/autoscaler/_private/command_runner.py
+61 -31
View File
@@ -718,7 +718,49 @@ class DockerCommandRunner(CommandRunnerInterface):
return string
def run_init(self, *, as_head, file_mounts):
def _check_if_container_restart_is_needed(
self, image: str, cleaned_bind_mounts: Dict[str, str]) -> bool:
re_init_required = False
running_image = self.run(
check_docker_image(self.container_name),
with_output=True,
run_env="host").decode("utf-8").strip()
if running_image != image:
cli_logger.error(
"A container with name {} is running image {} instead " +
"of {} (which was provided in the YAML)", self.container_name,
running_image, image)
mounts = self.run(
check_bind_mounts_cmd(self.container_name),
with_output=True,
run_env="host").decode("utf-8").strip()
try:
active_mounts = json.loads(mounts)
active_remote_mounts = {
mnt["Destination"].strip("/")
for mnt in active_mounts
}
# Ignore ray bootstrap files.
requested_remote_mounts = {
self._docker_expand_user(remote).strip("/")
for remote in cleaned_bind_mounts.keys()
}
unfulfilled_mounts = (
requested_remote_mounts - active_remote_mounts)
if unfulfilled_mounts:
re_init_required = True
cli_logger.warning(
"This Docker Container is already running. "
"Restarting the Docker container on "
"this node to pick up the following file_mounts {}",
unfulfilled_mounts)
except json.JSONDecodeError:
cli_logger.verbose(
"Unable to check if file_mounts specified in the YAML "
"differ from those on the running container.")
return re_init_required
def run_init(self, *, as_head, file_mounts, sync_run_yet):
BOOTSTRAP_MOUNTS = [
"~/ray_bootstrap_config.yaml", "~/ray_bootstrap_key.pem"
]
@@ -740,7 +782,17 @@ class DockerCommandRunner(CommandRunnerInterface):
for mnt in BOOTSTRAP_MOUNTS:
cleaned_bind_mounts.pop(mnt, None)
if not self._check_container_status():
docker_run_executed = False
container_running = self._check_container_status()
requires_re_init = False
if container_running:
requires_re_init = self._check_if_container_restart_is_needed(
image, cleaned_bind_mounts)
if requires_re_init:
self.run(f"docker stop {self.container_name}", run_env="host")
if (not container_running) or requires_re_init:
# Get home directory
image_env = self.ssh_command_runner.run(
"docker inspect -f '{{json .Config.Env}}' " + image,
@@ -760,39 +812,16 @@ class DockerCommandRunner(CommandRunnerInterface):
self._configure_runtime() + self._auto_configure_shm(),
self.ssh_command_runner.cluster_name, home_directory)
self.run(start_command, run_env="host")
else:
running_image = self.run(
check_docker_image(self.container_name),
with_output=True,
run_env="host").decode("utf-8").strip()
if running_image != image:
logger.error(f"A container with name {self.container_name} " +
f"is running image {running_image} instead " +
f"of {image} (which was provided in the YAML")
mounts = self.run(
check_bind_mounts_cmd(self.container_name),
with_output=True,
run_env="host").decode("utf-8").strip()
try:
active_mounts = json.loads(mounts)
active_remote_mounts = [
mnt["Destination"] for mnt in active_mounts
]
# Ignore ray bootstrap files.
for remote, local in cleaned_bind_mounts.items():
remote = self._docker_expand_user(remote)
if remote not in active_remote_mounts:
cli_logger.error(
"Please ray stop & restart cluster to "
f"allow mount {remote}:{local} to take hold")
except json.JSONDecodeError:
cli_logger.verbose(
"Unable to check if file_mounts specified in the YAML "
"differ from those on the running container.")
docker_run_executed = True
# Explicitly copy in ray bootstrap files.
for mount in BOOTSTRAP_MOUNTS:
if mount in file_mounts:
if not sync_run_yet:
# NOTE(ilr) This rsync is needed because when starting from
# a stopped instance, /tmp may be deleted and `run_init`
# is called before the first `file_sync` happens
self.run_rsync_up(file_mounts[mount], mount)
self.ssh_command_runner.run(
"docker cp {src} {container}:{dst}".format(
src=os.path.join(
@@ -801,6 +830,7 @@ class DockerCommandRunner(CommandRunnerInterface):
container=self.container_name,
dst=self._docker_expand_user(mount)))
self.initialized = True
return docker_run_executed
def _configure_runtime(self):
if self.docker_config.get("disable_automatic_runtime_detection"):
python/ray/autoscaler/_private/updater.py
+8 -3
View File
@@ -292,8 +292,12 @@ class NodeUpdater:
if node_tags.get(TAG_RAY_RUNTIME_CONFIG) == self.runtime_hash:
# When resuming from a stopped instance the runtime_hash may be the
# same, but the container will not be started.
self.cmd_runner.run_init(
as_head=self.is_head_node, file_mounts=self.file_mounts)
init_required = self.cmd_runner.run_init(
as_head=self.is_head_node,
file_mounts=self.file_mounts,
sync_run_yet=False)
if init_required:
node_tags[TAG_RAY_RUNTIME_CONFIG] += "-invalidate"
# runtime_hash will only change whenever the user restarts
# or updates their cluster with `get_or_create_head_node`
@@ -371,7 +375,8 @@ class NodeUpdater:
_numbered=("[]", 5, NUM_SETUP_STEPS)):
self.cmd_runner.run_init(
as_head=self.is_head_node,
file_mounts=self.file_mounts)
file_mounts=self.file_mounts,
sync_run_yet=True)
if self.setup_commands:
with cli_logger.group(
"Running setup commands",
python/ray/autoscaler/command_runner.py
+6 -1
View File
@@ -75,11 +75,16 @@ class CommandRunnerInterface:
"""Return the command the user can use to open a shell."""
raise NotImplementedError
def run_init(self, *, as_head: bool, file_mounts: Dict[str, str]) -> None:
def run_init(self, *, as_head: bool, file_mounts: Dict[str, str],
sync_run_yet: bool) -> Optional[bool]:
"""Used to run extra initialization commands.
Args:
as_head (bool): Run as head image or worker.
file_mounts (dict): Files to copy to the head and worker nodes.
sync_run_yet (bool): Whether sync has been run yet.
Returns:
optional (bool): Whether initialization was run.
"""
pass
python/ray/tests/test_autoscaler.py
+252 -10
View File
@@ -1,3 +1,4 @@
import json
import os
import shutil
from subprocess import CalledProcessError
@@ -396,9 +397,9 @@ class AutoscalingTest(unittest.TestCase):
self.provider = MockProvider()
runner = MockProcessRunner()
runner.respond_to_call("json .Mounts", ["[]"])
# Two initial calls to docker cp, one before run, two final calls to cp
# Two initial calls to docker cp, + 2 more calls during run_init
runner.respond_to_call(".State.Running",
["false", "false", "false", "true", "true"])
["false", "false", "false", "false"])
runner.respond_to_call("json .Config.Env", ["[]"])
commands.get_or_create_head_node(
SMALL_CLUSTER,
@@ -428,6 +429,171 @@ class AutoscalingTest(unittest.TestCase):
f"docker cp {docker_mount_prefix}/~/ray_bootstrap_config.yaml"
runner.assert_has_call("1.2.3.4", pattern=pattern_to_assert)
@unittest.skipIf(sys.platform == "win32", "Failing on Windows.")
def testGetOrCreateHeadNodeFromStopped(self):
self.testGetOrCreateHeadNode()
self.provider.cache_stopped = True
existing_nodes = self.provider.non_terminated_nodes({})
assert len(existing_nodes) == 1
self.provider.terminate_node(existing_nodes[0])
config_path = self.write_config(SMALL_CLUSTER)
runner = MockProcessRunner()
runner.respond_to_call("json .Mounts", ["[]"])
# Two initial calls to docker cp, + 2 more calls during run_init
runner.respond_to_call(".State.Running",
["false", "false", "false", "false"])
runner.respond_to_call("json .Config.Env", ["[]"])
commands.get_or_create_head_node(
SMALL_CLUSTER,
printable_config_file=config_path,
no_restart=False,
restart_only=False,
yes=True,
override_cluster_name=None,
_provider=self.provider,
_runner=runner)
self.waitForNodes(1)
# Init & Setup commands msut be run for Docker!
runner.assert_has_call("1.2.3.4", "init_cmd")
runner.assert_has_call("1.2.3.4", "head_setup_cmd")
runner.assert_has_call("1.2.3.4", "start_ray_head")
self.assertEqual(self.provider.mock_nodes[0].node_type, None)
runner.assert_has_call("1.2.3.4", pattern="docker run")
docker_mount_prefix = get_docker_host_mount_location(
SMALL_CLUSTER["cluster_name"])
runner.assert_not_has_call(
"1.2.3.4",
pattern=f"-v {docker_mount_prefix}/~/ray_bootstrap_config")
runner.assert_has_call(
"1.2.3.4",
pattern=f"docker cp {docker_mount_prefix}/~/ray_bootstrap_key.pem")
pattern_to_assert = \
f"docker cp {docker_mount_prefix}/~/ray_bootstrap_config.yaml"
runner.assert_has_call("1.2.3.4", pattern=pattern_to_assert)
# This next section of code ensures that the following order of
# commands are executed:
# 1. mkdir -p {docker_mount_prefix}
# 2. rsync bootstrap files
# 3. docker cp bootstrap files into container
commands_with_mount = [
(i, cmd) for i, cmd in enumerate(runner.command_history())
if docker_mount_prefix in cmd
]
rsync_commands = [x for x in commands_with_mount if "rsync" in x[1]]
docker_cp_commands = [
x for x in commands_with_mount if "docker cp" in x[1]
]
first_mkdir = min(x[0] for x in commands_with_mount if "mkdir" in x[1])
for file_to_check in [
"ray_bootstrap_config.yaml", "ray_bootstrap_key.pem"
]:
first_rsync = min(x[0] for x in rsync_commands
if "ray_bootstrap_config.yaml" in x[1])
first_cp = min(
x[0] for x in docker_cp_commands if file_to_check in x[1])
assert first_mkdir < first_rsync
assert first_rsync < first_cp
@unittest.skipIf(sys.platform == "win32", "Failing on Windows.")
def testDockerFileMountsAdded(self):
config = copy.deepcopy(SMALL_CLUSTER)
config["file_mounts"] = {"source": "/dev/null"}
config_path = self.write_config(config)
self.provider = MockProvider()
runner = MockProcessRunner()
mounts = [{
"Type": "bind",
"Source": "/sys",
"Destination": "/sys",
"Mode": "ro",
"RW": False,
"Propagation": "rprivate"
}]
runner.respond_to_call("json .Mounts", [json.dumps(mounts)])
# Two initial calls to docker cp, +1 more call during run_init
runner.respond_to_call(".State.Running",
["false", "false", "true", "true"])
runner.respond_to_call("json .Config.Env", ["[]"])
commands.get_or_create_head_node(
config,
printable_config_file=config_path,
no_restart=False,
restart_only=False,
yes=True,
override_cluster_name=None,
_provider=self.provider,
_runner=runner)
self.waitForNodes(1)
runner.assert_has_call("1.2.3.4", "init_cmd")
runner.assert_has_call("1.2.3.4", "head_setup_cmd")
runner.assert_has_call("1.2.3.4", "start_ray_head")
self.assertEqual(self.provider.mock_nodes[0].node_type, None)
runner.assert_has_call("1.2.3.4", pattern="docker stop")
runner.assert_has_call("1.2.3.4", pattern="docker run")
docker_mount_prefix = get_docker_host_mount_location(
SMALL_CLUSTER["cluster_name"])
runner.assert_not_has_call(
"1.2.3.4",
pattern=f"-v {docker_mount_prefix}/~/ray_bootstrap_config")
runner.assert_has_call(
"1.2.3.4",
pattern=f"docker cp {docker_mount_prefix}/~/ray_bootstrap_key.pem")
pattern_to_assert = \
f"docker cp {docker_mount_prefix}/~/ray_bootstrap_config.yaml"
runner.assert_has_call("1.2.3.4", pattern=pattern_to_assert)
def testDockerFileMountsRemoved(self):
config = copy.deepcopy(SMALL_CLUSTER)
config["file_mounts"] = {}
config_path = self.write_config(config)
self.provider = MockProvider()
runner = MockProcessRunner()
mounts = [{
"Type": "bind",
"Source": "/sys",
"Destination": "/sys",
"Mode": "ro",
"RW": False,
"Propagation": "rprivate"
}]
runner.respond_to_call("json .Mounts", [json.dumps(mounts)])
# Two initial calls to docker cp, +1 more call during run_init
runner.respond_to_call(".State.Running",
["false", "false", "true", "true"])
runner.respond_to_call("json .Config.Env", ["[]"])
commands.get_or_create_head_node(
config,
printable_config_file=config_path,
no_restart=False,
restart_only=False,
yes=True,
override_cluster_name=None,
_provider=self.provider,
_runner=runner)
self.waitForNodes(1)
runner.assert_has_call("1.2.3.4", "init_cmd")
runner.assert_has_call("1.2.3.4", "head_setup_cmd")
runner.assert_has_call("1.2.3.4", "start_ray_head")
self.assertEqual(self.provider.mock_nodes[0].node_type, None)
# We only removed amount from the YAML, no changes should happen.
runner.assert_not_has_call("1.2.3.4", pattern="docker stop")
runner.assert_not_has_call("1.2.3.4", pattern="docker run")
docker_mount_prefix = get_docker_host_mount_location(
SMALL_CLUSTER["cluster_name"])
runner.assert_not_has_call(
"1.2.3.4",
pattern=f"-v {docker_mount_prefix}/~/ray_bootstrap_config")
runner.assert_has_call(
"1.2.3.4",
pattern=f"docker cp {docker_mount_prefix}/~/ray_bootstrap_key.pem")
pattern_to_assert = \
f"docker cp {docker_mount_prefix}/~/ray_bootstrap_config.yaml"
runner.assert_has_call("1.2.3.4", pattern=pattern_to_assert)
@unittest.skipIf(sys.platform == "win32", "Failing on Windows.")
def testRsyncCommandWithDocker(self):
assert SMALL_CLUSTER["docker"]["container_name"]
@@ -1408,7 +1574,77 @@ class AutoscalingTest(unittest.TestCase):
runner.assert_has_call("172.0.0.1", "worker_setup_cmd")
runner.assert_has_call("172.0.0.1", "start_ray_worker")
def testSetupCommandsWithStoppedNodeCaching(self):
def testSetupCommandsWithStoppedNodeCachingNoDocker(self):
file_mount_dir = tempfile.mkdtemp()
config = SMALL_CLUSTER.copy()
del config["docker"]
config["file_mounts"] = {"/root/test-folder": file_mount_dir}
config["file_mounts_sync_continuously"] = True
config["min_workers"] = 1
config["max_workers"] = 1
config_path = self.write_config(config)
self.provider = MockProvider(cache_stopped=True)
runner = MockProcessRunner()
runner.respond_to_call("json .Config.Env", ["[]" for i in range(3)])
lm = LoadMetrics()
autoscaler = StandardAutoscaler(
config_path,
lm,
max_failures=0,
process_runner=runner,
update_interval_s=0)
autoscaler.update()
self.waitForNodes(1)
self.provider.finish_starting_nodes()
autoscaler.update()
self.waitForNodes(
1, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE})
runner.assert_has_call("172.0.0.0", "init_cmd")
runner.assert_has_call("172.0.0.0", "setup_cmd")
runner.assert_has_call("172.0.0.0", "worker_setup_cmd")
runner.assert_has_call("172.0.0.0", "start_ray_worker")
# Check the node was indeed reused
self.provider.terminate_node(0)
autoscaler.update()
self.waitForNodes(1)
runner.clear_history()
self.provider.finish_starting_nodes()
autoscaler.update()
self.waitForNodes(
1, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE})
runner.assert_not_has_call("172.0.0.0", "init_cmd")
runner.assert_not_has_call("172.0.0.0", "setup_cmd")
runner.assert_not_has_call("172.0.0.0", "worker_setup_cmd")
runner.assert_has_call("172.0.0.0", "start_ray_worker")
with open(f"{file_mount_dir}/new_file", "w") as f:
f.write("abcdefgh")
# Check that run_init happens when file_mounts have updated
self.provider.terminate_node(0)
autoscaler.update()
self.waitForNodes(1)
runner.clear_history()
self.provider.finish_starting_nodes()
autoscaler.update()
self.waitForNodes(
1, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE})
runner.assert_not_has_call("172.0.0.0", "init_cmd")
runner.assert_not_has_call("172.0.0.0", "setup_cmd")
runner.assert_not_has_call("172.0.0.0", "worker_setup_cmd")
runner.assert_has_call("172.0.0.0", "start_ray_worker")
runner.clear_history()
autoscaler.update()
runner.assert_not_has_call("172.0.0.0", "setup_cmd")
# We did not start any other nodes
runner.assert_not_has_call("172.0.0.1", " ")
def testSetupCommandsWithStoppedNodeCachingDocker(self):
# NOTE(ilr) Setup & Init commands **should** run with stopped nodes
# when Docker is in use.
file_mount_dir = tempfile.mkdtemp()
config = SMALL_CLUSTER.copy()
config["file_mounts"] = {"/root/test-folder": file_mount_dir}
@@ -1447,9 +1683,10 @@ class AutoscalingTest(unittest.TestCase):
autoscaler.update()
self.waitForNodes(
1, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE})
runner.assert_not_has_call("172.0.0.0", "init_cmd")
runner.assert_not_has_call("172.0.0.0", "setup_cmd")
runner.assert_not_has_call("172.0.0.0", "worker_setup_cmd")
# These all must happen when the node is stopped and resued
runner.assert_has_call("172.0.0.0", "init_cmd")
runner.assert_has_call("172.0.0.0", "setup_cmd")
runner.assert_has_call("172.0.0.0", "worker_setup_cmd")
runner.assert_has_call("172.0.0.0", "start_ray_worker")
runner.assert_has_call("172.0.0.0", "docker run")
@@ -1465,9 +1702,9 @@ class AutoscalingTest(unittest.TestCase):
autoscaler.update()
self.waitForNodes(
1, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE})
runner.assert_not_has_call("172.0.0.0", "init_cmd")
runner.assert_not_has_call("172.0.0.0", "setup_cmd")
runner.assert_not_has_call("172.0.0.0", "worker_setup_cmd")
runner.assert_has_call("172.0.0.0", "init_cmd")
runner.assert_has_call("172.0.0.0", "setup_cmd")
runner.assert_has_call("172.0.0.0", "worker_setup_cmd")
runner.assert_has_call("172.0.0.0", "start_ray_worker")
runner.assert_has_call("172.0.0.0", "docker run")
@@ -1480,12 +1717,13 @@ class AutoscalingTest(unittest.TestCase):
def testMultiNodeReuse(self):
config = SMALL_CLUSTER.copy()
# Docker re-runs setup commands when nodes are reused.
del config["docker"]
config["min_workers"] = 3
config["max_workers"] = 3
config_path = self.write_config(config)
self.provider = MockProvider(cache_stopped=True)
runner = MockProcessRunner()
runner.respond_to_call("json .Config.Env", ["[]" for i in range(13)])
lm = LoadMetrics()
autoscaler = StandardAutoscaler(
config_path,
@@ -1537,6 +1775,8 @@ class AutoscalingTest(unittest.TestCase):
config_path = self.write_config(config)
runner = MockProcessRunner()
runner.respond_to_call("json .Config.Env", ["[]" for i in range(4)])
runner.respond_to_call("command -v docker",
["docker" for _ in range(4)])
lm = LoadMetrics()
autoscaler = StandardAutoscaler(
config_path,
@@ -1565,6 +1805,8 @@ class AutoscalingTest(unittest.TestCase):
with open(os.path.join(file_mount_dir, "test.txt"), "wb") as temp_file:
temp_file.write("hello".encode())
runner.respond_to_call(".Config.Image", ["example" for _ in range(4)])
runner.respond_to_call(".State.Running", ["true" for _ in range(4)])
autoscaler.update()
self.waitForNodes(2)
self.provider.finish_starting_nodes()