From 7ad56826dbab8b718de47fbf2752ba0c81cbd843 Mon Sep 17 00:00:00 2001 From: Ian Rodney Date: Mon, 28 Dec 2020 18:56:28 -0800 Subject: [PATCH] [docker] Fix restart behavior with Docker (#12898) Co-authored-by: Richard Liaw Co-authored-by: ijrsvt --- .../ray/autoscaler/_private/command_runner.py | 92 +++--- python/ray/autoscaler/_private/updater.py | 11 +- python/ray/autoscaler/command_runner.py | 7 +- python/ray/tests/test_autoscaler.py | 262 +++++++++++++++++- 4 files changed, 327 insertions(+), 45 deletions(-) diff --git a/python/ray/autoscaler/_private/command_runner.py b/python/ray/autoscaler/_private/command_runner.py index f350ff1f3..5a6354078 100644 --- a/python/ray/autoscaler/_private/command_runner.py +++ b/python/ray/autoscaler/_private/command_runner.py @@ -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"): diff --git a/python/ray/autoscaler/_private/updater.py b/python/ray/autoscaler/_private/updater.py index 8616fbe84..7256d9046 100644 --- a/python/ray/autoscaler/_private/updater.py +++ b/python/ray/autoscaler/_private/updater.py @@ -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", diff --git a/python/ray/autoscaler/command_runner.py b/python/ray/autoscaler/command_runner.py index 151aafbed..b1be2ba0a 100644 --- a/python/ray/autoscaler/command_runner.py +++ b/python/ray/autoscaler/command_runner.py @@ -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 diff --git a/python/ray/tests/test_autoscaler.py b/python/ray/tests/test_autoscaler.py index 628c1b191..f859107e1 100644 --- a/python/ray/tests/test_autoscaler.py +++ b/python/ray/tests/test_autoscaler.py @@ -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()