From f1ace386db7396a7b22606c0143139b41008c192 Mon Sep 17 00:00:00 2001 From: Kai Fricke Date: Thu, 19 Nov 2020 21:17:17 +0100 Subject: [PATCH] [tune] detect docker and kubernetes syncers (#12108) Co-authored-by: Richard Liaw --- python/ray/tune/syncer.py | 43 +++++++++++++++++++++++++- python/ray/tune/tests/test_sync.py | 49 +++++++++++++++++++++++++++++- python/ray/tune/utils/callback.py | 17 ++++++----- 3 files changed, 100 insertions(+), 9 deletions(-) diff --git a/python/ray/tune/syncer.py b/python/ray/tune/syncer.py index e6180baa7..6e5ff23a0 100644 --- a/python/ray/tune/syncer.py +++ b/python/ray/tune/syncer.py @@ -1,4 +1,4 @@ -from typing import Any, Callable, Dict, List, TYPE_CHECKING, Union +from typing import Any, Callable, Dict, List, TYPE_CHECKING, Type, Union import distutils import logging @@ -10,6 +10,7 @@ from inspect import isclass from shlex import quote import ray +import yaml from ray import services from ray.tune import TuneError from ray.tune.callback import Callback @@ -448,3 +449,43 @@ class SyncerCallback(Callback): def on_checkpoint(self, iteration: int, trials: List["Trial"], trial: "Trial", checkpoint: Checkpoint, **info): self._sync_trial_checkpoint(trial, checkpoint) + + +def detect_sync_to_driver( + sync_to_driver: Union[None, bool, Type], + cluster_config_file: str = "~/ray_bootstrap_config.yaml"): + from ray.tune.integration.docker import DockerSyncer + from ray.tune.integration.kubernetes import NamespacedKubernetesSyncer + + if isinstance(sync_to_driver, Type): + return sync_to_driver + elif isinstance(sync_to_driver, bool) and sync_to_driver is False: + return sync_to_driver + + # Else: True or None. Auto-detect. + cluster_config_file = os.path.expanduser(cluster_config_file) + if not os.path.exists(cluster_config_file): + return sync_to_driver + + with open(cluster_config_file, "rt") as fp: + config = yaml.safe_load(fp.read()) + + if config.get("docker"): + logger.debug( + "Detected docker autoscaling environment. Using `DockerSyncer` " + "as sync client. If this is not correct or leads to errors, " + "please pass a `sync_to_driver` parameter in the `SyncConfig` to " + "`tune.run().` to manually configure syncing behavior.") + return DockerSyncer + + if config.get("provider", {}).get("type", "") == "kubernetes": + namespace = config["provider"].get("namespace", "ray") + logger.debug( + f"Detected Kubernetes autoscaling environment. Using " + f"`NamespacedKubernetesSyncer` with namespace `{namespace}` " + f"as sync client. If this is not correct or leads to errors, " + f"please pass a `sync_to_driver` parameter in the `SyncConfig` " + f"to `tune.run()` to manually configure syncing behavior..") + return NamespacedKubernetesSyncer(namespace) + + return sync_to_driver diff --git a/python/ray/tune/tests/test_sync.py b/python/ray/tune/tests/test_sync.py index a6159b890..1b8eae0a6 100644 --- a/python/ray/tune/tests/test_sync.py +++ b/python/ray/tune/tests/test_sync.py @@ -6,12 +6,15 @@ import tempfile import time import unittest from unittest.mock import patch +import yaml import ray from ray.rllib import _register_all from ray import tune -from ray.tune.syncer import CommandBasedClient +from ray.tune.integration.docker import DockerSyncer +from ray.tune.integration.kubernetes import KubernetesSyncer +from ray.tune.syncer import CommandBasedClient, detect_sync_to_driver class TestSyncFunctionality(unittest.TestCase): @@ -243,6 +246,50 @@ class TestSyncFunctionality(unittest.TestCase): sync_config=sync_config).trials self.assertEqual(mock_sync.call_count, 0) + def testSyncDetection(self): + kubernetes_conf = { + "provider": { + "type": "kubernetes", + "namespace": "test_ray" + } + } + docker_conf = { + "docker": { + "image": "bogus" + }, + "provider": { + "type": "aws" + } + } + aws_conf = {"provider": {"type": "aws"}} + + with tempfile.TemporaryDirectory() as dir: + kubernetes_file = os.path.join(dir, "kubernetes.yaml") + with open(kubernetes_file, "wt") as fp: + yaml.safe_dump(kubernetes_conf, fp) + + docker_file = os.path.join(dir, "docker.yaml") + with open(docker_file, "wt") as fp: + yaml.safe_dump(docker_conf, fp) + + aws_file = os.path.join(dir, "aws.yaml") + with open(aws_file, "wt") as fp: + yaml.safe_dump(aws_conf, fp) + + kubernetes_syncer = detect_sync_to_driver(None, kubernetes_file) + self.assertTrue(issubclass(kubernetes_syncer, KubernetesSyncer)) + self.assertEqual(kubernetes_syncer._namespace, "test_ray") + + docker_syncer = detect_sync_to_driver(None, docker_file) + self.assertTrue(issubclass(docker_syncer, DockerSyncer)) + + aws_syncer = detect_sync_to_driver(None, aws_file) + self.assertEqual(aws_syncer, None) + + # Should still return DockerSyncer, since it was passed explicitly + syncer = detect_sync_to_driver(DockerSyncer, kubernetes_file) + self.assertTrue(issubclass(syncer, DockerSyncer)) + if __name__ == "__main__": import pytest diff --git a/python/ray/tune/utils/callback.py b/python/ray/tune/utils/callback.py index d74b316a2..b2f5fcf71 100644 --- a/python/ray/tune/utils/callback.py +++ b/python/ray/tune/utils/callback.py @@ -1,11 +1,11 @@ -import logging -import os from typing import List, Optional +import logging +import os + from ray.tune.callback import Callback -from ray.tune.syncer import SyncConfig -from ray.tune.logger import CSVLoggerCallback, CSVLogger, \ - LoggerCallback, \ +from ray.tune.syncer import SyncConfig, detect_sync_to_driver +from ray.tune.logger import CSVLoggerCallback, CSVLogger, LoggerCallback, \ JsonLoggerCallback, JsonLogger, LegacyLoggerCallback, Logger, \ TBXLoggerCallback, TBXLogger from ray.tune.syncer import SyncerCallback @@ -84,8 +84,11 @@ def create_default_callbacks(callbacks: Optional[List[Callback]], # If no SyncerCallback was found, add if not has_syncer_callback and os.environ.get( "TUNE_DISABLE_AUTO_CALLBACK_SYNCER", "0") != "1": - syncer_callback = SyncerCallback( - sync_function=sync_config.sync_to_driver) + + # Detect Docker and Kubernetes environments + _sync_to_driver = detect_sync_to_driver(sync_config.sync_to_driver) + + syncer_callback = SyncerCallback(sync_function=_sync_to_driver) callbacks.append(syncer_callback) syncer_index = len(callbacks) - 1