wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-09-09 11:32:43 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix abstraction violations in command_runner interface (#10715)

Browse Source 
* Fix abstraction violations in command_runner interface

* user guide

* lint

* breaking abstraction in commands

* extra initialization commands

* more cleanup

* small fixes

* fix test_integration_kubernetes.py

* lint

Co-authored-by: root <root@ip-172-31-28-155.us-west-2.compute.internal>
Co-authored-by: Ameer Haj Ali <ameerhajali@Ameers-MacBook-Pro.local>
This commit is contained in:
Ameer Haj Ali
2020-09-14 20:28:38 -07:00
committed by GitHub
co-authored by root Ameer Haj Ali
parent 1a5cfe0b79
commit 6edacb22b8
6 changed files with 72 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/ray/tune/integration/kubernetes.py
+5 -23
View File
@@ -7,15 +7,11 @@ from ray.tune.syncer import NodeSyncer
from ray.tune.sync_client import SyncClient
def NamespacedKubernetesSyncer(namespace, use_rsync=False):
def NamespacedKubernetesSyncer(namespace):
"""Wrapper to return a ``KubernetesSyncer`` for a Kubernetes namespace.
Args:
namespace (str): Kubernetes namespace.
use_rsync (bool): Use ``rsync`` if True or ``kubectl cp``
if False. If True, ``rsync`` will need to be
installed in the Kubernetes pods for this to work.
If False, ``tar`` will need to be installed instead.
Returns: A ``KubernetesSyncer`` class to be passed to ``tune.run()``.
@@ -31,7 +27,6 @@ def NamespacedKubernetesSyncer(namespace, use_rsync=False):
class _NamespacedKubernetesSyncer(KubernetesSyncer):
_namespace = namespace
_use_rsync = use_rsync
return _NamespacedKubernetesSyncer
@@ -49,7 +44,6 @@ class KubernetesSyncer(NodeSyncer):
"""
_namespace = "ray"
_use_rsync = False
def __init__(self, local_dir, remote_dir, sync_client=None):
self.local_ip = services.get_node_ip_address()
@@ -58,8 +52,7 @@ class KubernetesSyncer(NodeSyncer):
self.worker_node = None
sync_client = sync_client or KubernetesSyncClient(
namespace=self.__class__._namespace,
use_rsync=self.__class__._use_rsync)
namespace=self.__class__._namespace)
super(NodeSyncer, self).__init__(local_dir, remote_dir, sync_client)
@@ -97,18 +90,13 @@ class KubernetesSyncClient(SyncClient):
Args:
namespace (str): Namespace in which the pods live.
use_rsync (bool): Use ``rsync`` if True or ``kubectl cp``
if False. If True, ``rsync`` will need to be
installed in the Kubernetes pods for this to work.
If False, ``tar`` will need to be installed instead.
process_runner: How commands should be called.
Defaults to ``subprocess``.
"""
def __init__(self, namespace, use_rsync=False, process_runner=subprocess):
def __init__(self, namespace, process_runner=subprocess):
self.namespace = namespace
self.use_rsync = use_rsync
self._process_runner = process_runner
self._command_runners = {}
@@ -141,10 +129,7 @@ class KubernetesSyncClient(SyncClient):
target_dir += "/" if not target_dir.endswith("/") else ""
command_runner = self._get_command_runner(target_node)
if self.use_rsync:
command_runner.run_rsync_up(source, target_dir)
else:
command_runner.run_cp_up(source, target_dir)
command_runner.run_rsync_up(source, target_dir)
return True
def sync_down(self, source, target):
@@ -156,10 +141,7 @@ class KubernetesSyncClient(SyncClient):
target += "/" if not target.endswith("/") else ""
command_runner = self._get_command_runner(source_node)
if self.use_rsync:
command_runner.run_rsync_down(source_dir, target)
else:
command_runner.run_cp_down(source_dir, target)
command_runner.run_rsync_down(source_dir, target)
return True
def delete(self, target):
python/ray/tune/tests/test_integration_kubernetes.py
+4 -40
View File
@@ -33,11 +33,10 @@ class _MockLookup:
return self.ip_to_node[ip]
def _create_mock_syncer(namespace, lookup, use_rsync, process_runner, local_ip,
local_dir, remote_dir):
def _create_mock_syncer(namespace, lookup, process_runner, local_ip, local_dir,
remote_dir):
class _MockSyncer(KubernetesSyncer):
_namespace = namespace
_use_rsync = use_rsync
_get_kubernetes_node_by_ip = lookup
def __init__(self, local_dir, remote_dir, sync_client):
@@ -54,9 +53,7 @@ def _create_mock_syncer(namespace, lookup, use_rsync, process_runner, local_ip,
local_dir,
remote_dir,
sync_client=KubernetesSyncClient(
namespace=namespace,
use_rsync=use_rsync,
process_runner=process_runner))
namespace=namespace, process_runner=process_runner))
class KubernetesIntegrationTest(unittest.TestCase):
@@ -74,42 +71,9 @@ class KubernetesIntegrationTest(unittest.TestCase):
def tearDown(self):
pass
def testKubernetesCpUpDown(self):
syncer = _create_mock_syncer(
self.namespace, self.lookup, False, self.process_runner,
self.lookup.get_ip("head"), self.local_dir, self.remote_dir)
syncer.set_worker_ip(self.lookup.get_ip("w1"))
# Test sync up. Should add / to the dirs and call kubectl cp
syncer.sync_up()
self.assertEqual(self.process_runner.history[-1], [
"kubectl", "-n", self.namespace, "cp", self.local_dir + "/",
"{}/{}:{}".format(self.namespace, "w1", self.remote_dir + "/")
])
# Test sync down.
syncer.sync_down()
self.assertEqual(self.process_runner.history[-1], [
"kubectl", "-n", self.namespace, "cp", "{}/{}:{}".format(
self.namespace,
"w1",
self.remote_dir + "/",
), self.local_dir + "/"
])
# Sync to same node should be ignored
syncer.set_worker_ip(self.lookup.get_ip("head"))
syncer.sync_up()
self.assertTrue(len(self.process_runner.history) == 2)
syncer.sync_down()
self.assertTrue(len(self.process_runner.history) == 2)
def testKubernetesRsyncUpDown(self):
syncer = _create_mock_syncer(
self.namespace, self.lookup, True, self.process_runner,
self.namespace, self.lookup, self.process_runner,
self.lookup.get_ip("head"), self.local_dir, self.remote_dir)
syncer.set_worker_ip(self.lookup.get_ip("w1"))