[tune] detect docker and kubernetes syncers

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 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,32 @@ def on_trial_complete(self, iteration: int, trials: List["Trial"],
     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", None):
+        return DockerSyncer
+
+    if config.get("provider", {}).get("type", "") == "kubernetes":
+        namespace = config["provider"].get("namespace", "ray")
+        return NamespacedKubernetesSyncer(namespace)
+
+    return sync_to_driver

python/ray/tune/tests/test_sync.py

@@ -6,12 +6,15 @@
 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 @@ def sync_func(source, target):
                 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

python/ray/tune/utils/callback.py

@@ -1,12 +1,12 @@
+from typing import List, Optional
+
 import logging
 import os
-from typing import List, Optional
 
 from ray.tune.callback import Callback
 from ray.tune.progress_reporter import TrialProgressCallback
-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
@@ -115,8 +115,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