From 81197e47c714a0df78c9c839d912d51fc03d9b12 Mon Sep 17 00:00:00 2001 From: Ujval Misra Date: Tue, 17 Dec 2019 05:25:16 -0800 Subject: [PATCH] [tune] Refactor syncer (#6496) * Refactor syncer and log_sync. * Fix documentation. * Remove delete from api * Rename to get_node_syncer --- python/ray/tune/log_sync.py | 107 ----------- python/ray/tune/logger.py | 6 +- python/ray/tune/sync_client.py | 203 +++++++++++++++++++++ python/ray/tune/syncer.py | 279 ++++++++++++----------------- python/ray/tune/tests/test_sync.py | 8 +- 5 files changed, 320 insertions(+), 283 deletions(-) delete mode 100644 python/ray/tune/log_sync.py create mode 100644 python/ray/tune/sync_client.py diff --git a/python/ray/tune/log_sync.py b/python/ray/tune/log_sync.py deleted file mode 100644 index a7d2be140..000000000 --- a/python/ray/tune/log_sync.py +++ /dev/null @@ -1,107 +0,0 @@ -from __future__ import absolute_import -from __future__ import division -from __future__ import print_function - -import distutils.spawn -import logging - -try: # py3 - from shlex import quote -except ImportError: # py2 - from pipes import quote - -import ray -from ray.tune.cluster_info import get_ssh_key, get_ssh_user - -logger = logging.getLogger(__name__) -_log_sync_warned = False - - -def log_sync_template(options=""): - """Syncs the local_dir between driver and worker if possible. - - Requires ray cluster to be started with the autoscaler. Also requires - rsync to be installed. - - Args: - options (str): Addtional rsync options. - - Returns: - Sync template with source and target parameters. - """ - if not distutils.spawn.find_executable("rsync"): - logger.error("Log sync requires rsync to be installed.") - return - global _log_sync_warned - ssh_key = get_ssh_key() - if ssh_key is None: - if not _log_sync_warned: - logger.debug("Log sync requires cluster to be setup with " - "`ray up`.") - _log_sync_warned = True - return - - rsh = "ssh -i {ssh_key} -o ConnectTimeout=120s -o StrictHostKeyChecking=no" - rsh = rsh.format(ssh_key=quote(ssh_key)) - template = """rsync {options} -savz -e "{rsh}" {{source}} {{target}}""" - return template.format(options=options, rsh=rsh) - - -class NodeSyncMixin(object): - # TODO(ujvl): Refactor this code. - """Mixin for syncing files to/from a remote dir to a local dir.""" - - def __init__(self): - assert hasattr(self, "_remote_dir"), "Mixin not mixed with Syncer." - self.local_ip = ray.services.get_node_ip_address() - self.worker_ip = None - - def set_worker_ip(self, worker_ip): - """Set the worker ip to sync logs from.""" - self.worker_ip = worker_ip - - def has_remote_target(self): - """Returns whether the Syncer has a remote target.""" - if not self.worker_ip: - logger.debug("Worker IP unknown, skipping log sync for %s", - self._local_dir) - return False - if self.worker_ip == self.local_ip: - logger.debug("Worker IP is local IP, skipping log sync for %s", - self._local_dir) - return False - return True - - def sync_up_if_needed(self): - if not self.has_remote_target(): - return True - super(NodeSyncMixin, self).sync_up() - - def sync_down_if_needed(self): - if not self.has_remote_target(): - return True - super(NodeSyncMixin, self).sync_down() - - def sync_down(self): - if not self.has_remote_target(): - return True - return super(NodeSyncMixin, self).sync_down() - - def sync_up(self): - if not self.has_remote_target(): - return True - return super(NodeSyncMixin, self).sync_up() - - @property - def _remote_path(self): - ssh_user = get_ssh_user() - global _log_sync_warned - if not self.has_remote_target(): - return - if ssh_user is None: - if not _log_sync_warned: - logger.error("Log sync requires cluster to be setup with " - "`ray up`.") - _log_sync_warned = True - return - return "{}@{}:{}/".format(ssh_user, self.worker_ip, self._remote_dir) diff --git a/python/ray/tune/logger.py b/python/ray/tune/logger.py index 7b61e6f2e..5ee4ea3f5 100644 --- a/python/ray/tune/logger.py +++ b/python/ray/tune/logger.py @@ -14,7 +14,7 @@ import numpy as np import ray.cloudpickle as cloudpickle from ray.tune.util import flatten_dict -from ray.tune.syncer import get_log_syncer +from ray.tune.syncer import get_node_syncer from ray.tune.result import (NODE_IP, TRAINING_ITERATION, TIME_TOTAL_S, TIMESTEPS_TOTAL, EXPR_PARAM_FILE, EXPR_PARAM_PICKLE_FILE, EXPR_PROGRESS_FILE, @@ -385,7 +385,7 @@ class UnifiedLogger(Logger): loggers (list): List of logger creators. Defaults to CSV, Tensorboard, and JSON loggers. sync_function (func|str): Optional function for syncer to run. - See ray/python/ray/tune/log_sync.py + See ray/python/ray/tune/syncer.py """ def __init__(self, @@ -411,7 +411,7 @@ class UnifiedLogger(Logger): except Exception as exc: logger.warning("Could not instantiate %s: %s.", cls.__name__, str(exc)) - self._log_syncer = get_log_syncer( + self._log_syncer = get_node_syncer( self.logdir, remote_dir=self.logdir, sync_function=self._sync_function) diff --git a/python/ray/tune/sync_client.py b/python/ray/tune/sync_client.py new file mode 100644 index 000000000..3f6d5dba6 --- /dev/null +++ b/python/ray/tune/sync_client.py @@ -0,0 +1,203 @@ +from __future__ import absolute_import +from __future__ import division +from __future__ import print_function + +import distutils +import distutils.spawn +import logging +import subprocess +import tempfile +import types + +try: # py3 + from shlex import quote +except ImportError: # py2 + from pipes import quote +from ray.tune.error import TuneError + +logger = logging.getLogger(__name__) + +S3_PREFIX = "s3://" +GS_PREFIX = "gs://" +ALLOWED_REMOTE_PREFIXES = (S3_PREFIX, GS_PREFIX) + +noop_template = ": {target}" # noop in bash + + +def noop(*args): + return + + +def get_sync_client(sync_function): + """Returns a sync client. + + Args: + sync_function (str|function): Sync function. + + Raises: + ValueError if sync_function is malformed. + """ + if isinstance(sync_function, types.FunctionType): + client_cls = FunctionBasedClient + elif isinstance(sync_function, str): + client_cls = CommandBasedClient + else: + raise ValueError("Sync function {} must be string or function".format( + sync_function)) + return client_cls(sync_function, sync_function) + + +def get_cloud_sync_client(remote_path): + """Returns a CommandBasedClient that can sync to/from remote storage. + + Args: + remote_path (str): Path to remote storage (S3 or GS). + + Raises: + ValueError if malformed remote_dir. + """ + if remote_path.startswith(S3_PREFIX): + if not distutils.spawn.find_executable("aws"): + raise ValueError( + "Upload uri starting with '{}' requires awscli tool" + " to be installed".format(S3_PREFIX)) + template = "aws s3 sync {source} {target}" + elif remote_path.startswith(GS_PREFIX): + if not distutils.spawn.find_executable("gsutil"): + raise ValueError( + "Upload uri starting with '{}' requires gsutil tool" + " to be installed".format(GS_PREFIX)) + template = "gsutil rsync -r {source} {target}" + else: + raise ValueError("Upload uri must start with one of: {}" + "".format(ALLOWED_REMOTE_PREFIXES)) + return CommandBasedClient(template, template) + + +class SyncClient(object): + """Client interface for interacting with remote storage options.""" + + def sync_up(self, source, target): + """Syncs up from source to target. + + Args: + source (str): Source path. + target (str): Target path. + + Returns: + True if sync initiation successful, False otherwise. + """ + raise NotImplementedError + + def sync_down(self, source, target): + """Syncs down from source to target. + + Args: + source (str): Source path. + target (str): Target path. + + Returns: + True if sync initiation successful, False otherwise. + """ + raise NotImplementedError + + def wait(self): + """Waits for current sync to complete, if asynchronously started.""" + pass + + def reset(self): + """Resets state.""" + pass + + +class FunctionBasedClient(SyncClient): + def __init__(self, sync_up_func, sync_down_func): + self.sync_up_func = sync_up_func + self.sync_down_func = sync_down_func + + def sync_up(self, source, target): + self.sync_up_func(source, target) + return True + + def sync_down(self, source, target): + self.sync_down_func(source, target) + return True + + +NOOP = FunctionBasedClient(noop, noop) + + +class CommandBasedClient(SyncClient): + def __init__(self, sync_up_template, sync_down_template): + """Syncs between two directories with the given command. + + Arguments: + sync_up_template (str): A runnable string template; needs to + include replacement fields '{source}' and '{target}'. + sync_down_template (str): A runnable string template; needs to + include replacement fields '{source}' and '{target}'. + """ + self._validate_sync_string(sync_up_template) + self._validate_sync_string(sync_down_template) + self.sync_up_template = sync_up_template + self.sync_down_template = sync_down_template + self.logfile = None + self.cmd_process = None + + def set_logdir(self, logdir): + """Sets the directory to log sync execution output in. + + Args: + logdir (str): Log directory. + """ + self.logfile = tempfile.NamedTemporaryFile( + prefix="log_sync", dir=logdir, suffix=".log", delete=False) + + def sync_up(self, source, target): + return self._execute(self.sync_up_template, source, target) + + def sync_down(self, source, target): + return self._execute(self.sync_down_template, source, target) + + def wait(self): + if self.cmd_process: + _, error_msg = self.cmd_process.communicate() + error_msg = error_msg.decode("ascii") + code = self.cmd_process.returncode + self.cmd_process = None + if code != 0: + raise TuneError("Sync error ({}): {}".format(code, error_msg)) + + def reset(self): + if self.is_running: + logger.warning("Sync process still running but resetting anyways.") + self.cmd_process = None + + @property + def is_running(self): + """Returns whether a sync process is running.""" + if self.cmd_process: + self.cmd_process.poll() + return self.cmd_process.returncode is None + return False + + def _execute(self, sync_template, source, target): + """Executes sync_template on source and target.""" + if self.is_running: + logger.warning("Last sync client cmd still in progress, skipping.") + return False + final_cmd = sync_template.format( + source=quote(source), target=quote(target)) + logger.debug("Running sync: {}".format(final_cmd)) + self.cmd_process = subprocess.Popen( + final_cmd, shell=True, stderr=subprocess.PIPE, stdout=self.logfile) + return True + + @staticmethod + def _validate_sync_string(sync_string): + if not isinstance(sync_string, str): + raise ValueError("{} is not a string.".format(sync_string)) + if "{source}" not in sync_string: + raise ValueError("Sync template missing '{source}'.") + if "{target}" not in sync_string: + raise ValueError("Sync template missing '{target}'.") diff --git a/python/ray/tune/syncer.py b/python/ray/tune/syncer.py index fde57b095..80d482b60 100644 --- a/python/ray/tune/syncer.py +++ b/python/ray/tune/syncer.py @@ -5,26 +5,23 @@ from __future__ import print_function import distutils import logging import os -import subprocess -import tempfile import time -import types try: # py3 from shlex import quote except ImportError: # py2 from pipes import quote -from ray.tune.error import TuneError -from ray.tune.log_sync import log_sync_template, NodeSyncMixin +from ray import services +from ray.tune.cluster_info import get_ssh_key, get_ssh_user +from ray.tune.sync_client import (CommandBasedClient, get_sync_client, + get_cloud_sync_client, NOOP) logger = logging.getLogger(__name__) -S3_PREFIX = "s3://" -GS_PREFIX = "gs://" -ALLOWED_REMOTE_PREFIXES = (S3_PREFIX, GS_PREFIX) SYNC_PERIOD = 300 +_log_sync_warned = False _syncers = {} @@ -33,127 +30,34 @@ def wait_for_sync(): syncer.wait() -class SyncClient(object): - def sync_up(self, source, target): - """Sync up from source to target. +def log_sync_template(options=""): + """Template enabling syncs between driver and worker when possible. + Requires ray cluster to be started with the autoscaler. Also requires + rsync to be installed. - Args: - source (str): Source path. - target (str): Target path. + Args: + options (str): Additional rsync options. - Returns: - True if sync initiation successful, False otherwise. - """ - raise NotImplementedError + Returns: + Sync template with source and target parameters. None if rsync + unavailable. + """ + if not distutils.spawn.find_executable("rsync"): + logger.error("Log sync requires rsync to be installed.") + return None + global _log_sync_warned + ssh_key = get_ssh_key() + if ssh_key is None: + if not _log_sync_warned: + logger.debug("Log sync requires cluster to be setup with " + "`ray up`.") + _log_sync_warned = True + return None - def sync_down(self, source, target): - """Sync down from source to target. - - Args: - source (str): Source path. - target (str): Target path. - - Returns: - True if sync initiation successful, False otherwise. - """ - raise NotImplementedError - - def wait(self): - """Wait for current sync to complete, if asynchronously started.""" - pass - - def reset(self): - """Resets state.""" - pass - - -class FunctionBasedClient(SyncClient): - def __init__(self, sync_up_func, sync_down_func): - self.sync_up_func = sync_up_func - self.sync_down_func = sync_down_func - - def sync_up(self, source, target): - self.sync_up_func(source, target) - return True - - def sync_down(self, source, target): - self.sync_down_func(source, target) - return True - - -class CommandBasedClient(SyncClient): - def __init__(self, sync_up_template, sync_down_template): - """Syncs between two directories with the given command. - - Arguments: - sync_up_template (str): A runnable string template; needs to - include replacement fields '{source}' and '{target}'. - sync_down_template (str): A runnable string template; needs to - include replacement fields '{source}' and '{target}'. - """ - if not isinstance(sync_up_template, str): - raise ValueError("{} is not a string.".format(sync_up_template)) - if not isinstance(sync_down_template, str): - raise ValueError("{} is not a string.".format(sync_down_template)) - self._validate_sync_string(sync_up_template) - self._validate_sync_string(sync_down_template) - self.sync_up_template = sync_up_template - self.sync_down_template = sync_down_template - self.logfile = None - self.sync_process = None - - def set_logdir(self, logdir): - """Sets the directory to log sync execution output in. - - Args: - logdir: Log directory. - """ - self.logfile = tempfile.NamedTemporaryFile( - prefix="log_sync", dir=logdir, suffix=".log", delete=False) - - def sync_up(self, source, target): - return self.execute(self.sync_up_template, source, target) - - def sync_down(self, source, target): - return self.execute(self.sync_down_template, source, target) - - def execute(self, sync_template, source, target): - """Executes sync_template on source and target.""" - if self.sync_process: - self.sync_process.poll() - if self.sync_process.returncode is None: - logger.warning("Last sync is still in progress, skipping.") - return False - final_cmd = sync_template.format( - source=quote(source), target=quote(target)) - logger.debug("Running sync: {}".format(final_cmd)) - self.sync_process = subprocess.Popen( - final_cmd, shell=True, stderr=subprocess.PIPE, stdout=self.logfile) - return True - - def wait(self): - if self.sync_process: - _, error_msg = self.sync_process.communicate() - error_msg = error_msg.decode("ascii") - code = self.sync_process.returncode - self.sync_process = None - if code != 0: - raise TuneError("Sync error ({}): {}".format(code, error_msg)) - - def reset(self): - if self.sync_process: - logger.warning("Sync process still running but resetting anyways.") - self.sync_process = None - - @staticmethod - def _validate_sync_string(sync_string): - if "{source}" not in sync_string: - raise ValueError("Sync template missing '{source}'.") - if "{target}" not in sync_string: - raise ValueError("Sync template missing '{target}'.") - - -NOOP = FunctionBasedClient(lambda s, t: None, lambda s, t: None) + rsh = "ssh -i {ssh_key} -o ConnectTimeout=120s -o StrictHostKeyChecking=no" + rsh = rsh.format(ssh_key=quote(ssh_key)) + template = "rsync {options} -savz -e {rsh} {{source}} {{target}}" + return template.format(options=options, rsh=quote(rsh)) class Syncer(object): @@ -234,6 +138,76 @@ class Syncer(object): return self._remote_dir +class NodeSyncer(Syncer): + """Syncer for syncing files to/from a remote dir to a local dir.""" + + def __init__(self, local_dir, remote_dir, sync_client): + self.local_ip = services.get_node_ip_address() + self.worker_ip = None + super(NodeSyncer, self).__init__(local_dir, remote_dir, sync_client) + + def set_worker_ip(self, worker_ip): + """Sets the worker IP to sync logs from.""" + self.worker_ip = worker_ip + + def has_remote_target(self): + """Returns whether the Syncer has a remote target.""" + if not self.worker_ip: + logger.debug("Worker IP unknown, skipping sync for %s", + self._local_dir) + return False + if self.worker_ip == self.local_ip: + logger.debug("Worker IP is local IP, skipping sync for %s", + self._local_dir) + return False + return True + + def sync_up_if_needed(self): + if not self.has_remote_target(): + return True + return super(NodeSyncer, self).sync_up_if_needed() + + def sync_down_if_needed(self): + if not self.has_remote_target(): + return True + return super(NodeSyncer, self).sync_down_if_needed() + + def sync_up_to_new_location(self, worker_ip): + if worker_ip != self.worker_ip: + logger.debug("Setting new worker IP to %s", worker_ip) + self.set_worker_ip(worker_ip) + self.reset() + if not self.sync_up(): + logger.warning( + "Sync up to new location skipped. This should not occur.") + else: + logger.warning("Sync attempted to same IP %s.", worker_ip) + + def sync_up(self): + if not self.has_remote_target(): + return True + return super(NodeSyncer, self).sync_up() + + def sync_down(self): + if not self.has_remote_target(): + return True + return super(NodeSyncer, self).sync_down() + + @property + def _remote_path(self): + ssh_user = get_ssh_user() + global _log_sync_warned + if not self.has_remote_target(): + return None + if ssh_user is None: + if not _log_sync_warned: + logger.error("Syncer requires cluster to be setup with " + "`ray up`.") + _log_sync_warned = True + return None + return "{}@{}:{}/".format(ssh_user, self.worker_ip, self._remote_dir) + + def get_cloud_syncer(local_dir, remote_dir=None, sync_function=None): """Returns a Syncer. @@ -247,6 +221,9 @@ def get_cloud_syncer(local_dir, remote_dir=None, sync_function=None): remote_dir. If string, then it must be a string template for syncer to run. If not provided, it defaults to standard S3 or gsutil sync commands. + + Raises: + ValueError if malformed remote_dir. """ key = (local_dir, remote_dir) @@ -257,37 +234,18 @@ def get_cloud_syncer(local_dir, remote_dir=None, sync_function=None): _syncers[key] = Syncer(local_dir, remote_dir, NOOP) return _syncers[key] - client = _get_sync_client(sync_function) + client = get_sync_client(sync_function) if client: _syncers[key] = Syncer(local_dir, remote_dir, client) return _syncers[key] - - if remote_dir.startswith(S3_PREFIX): - if not distutils.spawn.find_executable("aws"): - raise TuneError( - "Upload uri starting with '{}' requires awscli tool" - " to be installed".format(S3_PREFIX)) - template = "aws s3 sync {source} {target}" - s3_client = CommandBasedClient(template, template) - _syncers[key] = Syncer(local_dir, remote_dir, s3_client) - elif remote_dir.startswith(GS_PREFIX): - if not distutils.spawn.find_executable("gsutil"): - raise TuneError( - "Upload uri starting with '{}' requires gsutil tool" - " to be installed".format(GS_PREFIX)) - template = "gsutil rsync -r {source} {target}" - gs_client = CommandBasedClient(template, template) - _syncers[key] = Syncer(local_dir, remote_dir, gs_client) - else: - raise TuneError("Upload uri must start with one of: {}" - "".format(ALLOWED_REMOTE_PREFIXES)) - + sync_client = get_cloud_sync_client(remote_dir) + _syncers[key] = Syncer(local_dir, remote_dir, sync_client) return _syncers[key] -def get_log_syncer(local_dir, remote_dir=None, sync_function=None): - """Returns a log Syncer. +def get_node_syncer(local_dir, remote_dir=None, sync_function=None): + """Returns a NodeSyncer. Args: local_dir (str): Source directory for syncing. @@ -303,7 +261,7 @@ def get_log_syncer(local_dir, remote_dir=None, sync_function=None): elif not remote_dir: sync_client = NOOP elif sync_function: - sync_client = _get_sync_client(sync_function) + sync_client = get_sync_client(sync_function) else: sync_up = log_sync_template() sync_down = log_sync_template(options="--remove-source-files") @@ -313,22 +271,5 @@ def get_log_syncer(local_dir, remote_dir=None, sync_function=None): else: sync_client = NOOP - class MixedSyncer(NodeSyncMixin, Syncer): - def __init__(self, *args, **kwargs): - Syncer.__init__(self, *args, **kwargs) - NodeSyncMixin.__init__(self) - - _syncers[key] = MixedSyncer(local_dir, remote_dir, sync_client) + _syncers[key] = NodeSyncer(local_dir, remote_dir, sync_client) return _syncers[key] - - -def _get_sync_client(sync_function): - if not sync_function: - return None - if isinstance(sync_function, types.FunctionType): - return FunctionBasedClient(sync_function, sync_function) - elif isinstance(sync_function, str): - return CommandBasedClient(sync_function, sync_function) - else: - raise ValueError("Sync function {} must be string or function".format( - sync_function)) diff --git a/python/ray/tune/tests/test_sync.py b/python/ray/tune/tests/test_sync.py index b103236ed..85c36470b 100644 --- a/python/ray/tune/tests/test_sync.py +++ b/python/ray/tune/tests/test_sync.py @@ -30,7 +30,7 @@ class TestSyncFunctionality(unittest.TestCase): ray.shutdown() _register_all() # re-register the evicted objects - @patch("ray.tune.syncer.S3_PREFIX", "test") + @patch("ray.tune.sync_client.S3_PREFIX", "test") def testNoUploadDir(self): """No Upload Dir is given.""" with self.assertRaises(AssertionError): @@ -45,7 +45,7 @@ class TestSyncFunctionality(unittest.TestCase): "sync_to_cloud": "echo {source} {target}" }).trials - @patch("ray.tune.syncer.S3_PREFIX", "test") + @patch("ray.tune.sync_client.S3_PREFIX", "test") def testCloudProperString(self): with self.assertRaises(ValueError): [trial] = tune.run( @@ -120,7 +120,7 @@ class TestSyncFunctionality(unittest.TestCase): "sync_to_driver": "ls {source}" }).trials - with patch.object(CommandBasedClient, "execute") as mock_fn: + with patch.object(CommandBasedClient, "_execute") as mock_fn: with patch("ray.services.get_node_ip_address") as mock_sync: mock_sync.return_value = "0.0.0.0" [trial] = tune.run( @@ -198,7 +198,7 @@ class TestSyncFunctionality(unittest.TestCase): def sync_func(source, target): pass - with patch.object(CommandBasedClient, "execute") as mock_sync: + with patch.object(CommandBasedClient, "_execute") as mock_sync: [trial] = tune.run( "__fake", name="foo",