We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[autoscaler] bash 4.4 does not support set -i
<!--
General questions should be asked on the mailing list ray-dev@googlegroups.com.

Before submitting an issue, please fill out the following form.
-->

### System information
- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**:
Ubuntu 17.10

- **Ray installed from (source or binary)**:
Source
- **Ray version**:
commit b8811cbe3418ab0d3ea10deaa54947d5bb26cecf

- **Python version**:
3.6
- **Exact command to reproduce**:
ray create_or_update example.yaml

<!--
You can obtain the Ray version with

python -c "import ray; print(ray.__version__)"
-->

### Describe the problem
<!-- Describe the problem clearly here. -->

As of bash 4.4, set -i is no longer accepted to create an interactive shell. Consider -t in stead.

### Source code / logs
<!-- Include any logs or source code that would be helpful to diagnose the problem. If including tracebacks, please include the full traceback. Large logs and files should be attached. Try to provide a reproducible test case that is the bare minimum necessary to generate the problem. -->


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `python/ray/autoscaler/updater.py`
Content:
```
1 from __future__ import absolute_import
2 from __future__ import division
3 from __future__ import print_function
4 
5 try:  # py3
6     from shlex import quote
7 except ImportError:  # py2
8     from pipes import quote
9 import logging
10 import os
11 import subprocess
12 import sys
13 import time
14 
15 from threading import Thread
16 
17 from ray.autoscaler.tags import TAG_RAY_NODE_STATUS, TAG_RAY_RUNTIME_CONFIG
18 from ray.autoscaler.log_timer import LogTimer
19 
20 logger = logging.getLogger(__name__)
21 
22 # How long to wait for a node to start, in seconds
23 NODE_START_WAIT_S = 300
24 SSH_CHECK_INTERVAL = 5
25 SSH_CONTROL_PATH = "/tmp/ray_ssh_sockets"
26 
27 
28 def get_default_ssh_options(private_key, connect_timeout):
29     OPTS = [
30         ("ConnectTimeout", "{}s".format(connect_timeout)),
31         ("StrictHostKeyChecking", "no"),
32         ("ControlMaster", "auto"),
33         ("ControlPath", "{}/%C".format(SSH_CONTROL_PATH)),
34         ("ControlPersist", "yes"),
35     ]
36 
37     return ["-i", private_key] + [
38         x for y in (["-o", "{}={}".format(k, v)] for k, v in OPTS) for x in y
39     ]
40 
41 
42 class NodeUpdater(object):
43     """A process for syncing files and running init commands on a node."""
44 
45     def __init__(self,
46                  node_id,
47                  provider_config,
48                  provider,
49                  auth_config,
50                  cluster_name,
51                  file_mounts,
52                  initialization_commands,
53                  setup_commands,
54                  runtime_hash,
55                  process_runner=subprocess,
56                  use_internal_ip=False):
57         self.daemon = True
58         self.process_runner = process_runner
59         self.node_id = node_id
60         self.use_internal_ip = (use_internal_ip or provider_config.get(
61             "use_internal_ips", False))
62         self.provider = provider
63         self.ssh_private_key = auth_config["ssh_private_key"]
64         self.ssh_user = auth_config["ssh_user"]
65         self.ssh_ip = None
66         self.file_mounts = {
67             remote: os.path.expanduser(local)
68             for remote, local in file_mounts.items()
69         }
70         self.initialization_commands = initialization_commands
71         self.setup_commands = setup_commands
72         self.runtime_hash = runtime_hash
73 
74     def get_caller(self, check_error):
75         if check_error:
76             return self.process_runner.call
77         else:
78             return self.process_runner.check_call
79 
80     def get_node_ip(self):
81         if self.use_internal_ip:
82             return self.provider.internal_ip(self.node_id)
83         else:
84             return self.provider.external_ip(self.node_id)
85 
86     def wait_for_ip(self, deadline):
87         while time.time() < deadline and \
88                 not self.provider.is_terminated(self.node_id):
89             logger.info("NodeUpdater: "
90                         "Waiting for IP of {}...".format(self.node_id))
91             ip = self.get_node_ip()
92             if ip is not None:
93                 return ip
94             time.sleep(10)
95 
96         return None
97 
98     def set_ssh_ip_if_required(self):
99         if self.ssh_ip is not None:
100             return
101 
102         # We assume that this never changes.
103         #   I think that's reasonable.
104         deadline = time.time() + NODE_START_WAIT_S
105         with LogTimer("NodeUpdater: {}: Got IP".format(self.node_id)):
106             ip = self.wait_for_ip(deadline)
107             assert ip is not None, "Unable to find IP of node"
108 
109         self.ssh_ip = ip
110 
111         # This should run before any SSH commands and therefore ensure that
112         #   the ControlPath directory exists, allowing SSH to maintain
113         #   persistent sessions later on.
114         with open("/dev/null", "w") as redirect:
115             self.get_caller(False)(
116                 ["mkdir", "-p", SSH_CONTROL_PATH],
117                 stdout=redirect,
118                 stderr=redirect)
119 
120             self.get_caller(False)(
121                 ["chmod", "0700", SSH_CONTROL_PATH],
122                 stdout=redirect,
123                 stderr=redirect)
124 
125     def run(self):
126         logger.info("NodeUpdater: "
127                     "{}: Updating to {}".format(self.node_id,
128                                                 self.runtime_hash))
129         try:
130             m = "{}: Applied config {}".format(self.node_id, self.runtime_hash)
131             with LogTimer("NodeUpdater: {}".format(m)):
132                 self.do_update()
133         except Exception as e:
134             error_str = str(e)
135             if hasattr(e, "cmd"):
136                 error_str = "(Exit Status {}) {}".format(
137                     e.returncode, " ".join(e.cmd))
138             logger.error("NodeUpdater: "
139                          "{}: Error updating {}".format(
140                              self.node_id, error_str))
141             self.provider.set_node_tags(self.node_id,
142                                         {TAG_RAY_NODE_STATUS: "update-failed"})
143             raise e
144 
145         self.provider.set_node_tags(
146             self.node_id, {
147                 TAG_RAY_NODE_STATUS: "up-to-date",
148                 TAG_RAY_RUNTIME_CONFIG: self.runtime_hash
149             })
150 
151         self.exitcode = 0
152 
153     def wait_for_ssh(self, deadline):
154         logger.info("NodeUpdater: "
155                     "{}: Waiting for SSH...".format(self.node_id))
156 
157         while time.time() < deadline and \
158                 not self.provider.is_terminated(self.node_id):
159             try:
160                 logger.debug("NodeUpdater: "
161                              "{}: Waiting for SSH...".format(self.node_id))
162                 self.ssh_cmd(
163                     "uptime",
164                     connect_timeout=5,
165                     redirect=open("/dev/null", "w"))
166                 return True
167 
168             except Exception as e:
169                 retry_str = str(e)
170                 if hasattr(e, "cmd"):
171                     retry_str = "(Exit Status {}): {}".format(
172                         e.returncode, " ".join(e.cmd))
173                 logger.debug("NodeUpdater: "
174                              "{}: SSH not up, retrying: {}".format(
175                                  self.node_id, retry_str))
176                 time.sleep(SSH_CHECK_INTERVAL)
177 
178         return False
179 
180     def do_update(self):
181         self.provider.set_node_tags(self.node_id,
182                                     {TAG_RAY_NODE_STATUS: "waiting-for-ssh"})
183 
184         deadline = time.time() + NODE_START_WAIT_S
185         self.set_ssh_ip_if_required()
186 
187         # Wait for SSH access
188         with LogTimer("NodeUpdater: " "{}: Got SSH".format(self.node_id)):
189             ssh_ok = self.wait_for_ssh(deadline)
190             assert ssh_ok, "Unable to SSH to node"
191 
192         # Rsync file mounts
193         self.provider.set_node_tags(self.node_id,
194                                     {TAG_RAY_NODE_STATUS: "syncing-files"})
195         for remote_path, local_path in self.file_mounts.items():
196             logger.info("NodeUpdater: "
197                         "{}: Syncing {} to {}...".format(
198                             self.node_id, local_path, remote_path))
199             assert os.path.exists(local_path), local_path
200             if os.path.isdir(local_path):
201                 if not local_path.endswith("/"):
202                     local_path += "/"
203                 if not remote_path.endswith("/"):
204                     remote_path += "/"
205 
206             m = "{}: Synced {} to {}".format(self.node_id, local_path,
207                                              remote_path)
208             with LogTimer("NodeUpdater {}".format(m)):
209                 self.ssh_cmd(
210                     "mkdir -p {}".format(os.path.dirname(remote_path)),
211                     redirect=open("/dev/null", "w"),
212                 )
213                 self.rsync_up(
214                     local_path, remote_path, redirect=open("/dev/null", "w"))
215 
216         # Run init commands
217         self.provider.set_node_tags(self.node_id,
218                                     {TAG_RAY_NODE_STATUS: "setting-up"})
219 
220         m = "{}: Initialization commands completed".format(self.node_id)
221         with LogTimer("NodeUpdater: {}".format(m)):
222             for cmd in self.initialization_commands:
223                 self.ssh_cmd(cmd, redirect=open("/dev/null", "w"))
224 
225         m = "{}: Setup commands completed".format(self.node_id)
226         with LogTimer("NodeUpdater: {}".format(m)):
227             for cmd in self.setup_commands:
228                 self.ssh_cmd(cmd, redirect=open("/dev/null", "w"))
229 
230     def rsync_up(self, source, target, redirect=None, check_error=True):
231         self.set_ssh_ip_if_required()
232         self.get_caller(check_error)(
233             [
234                 "rsync", "-e",
235                 " ".join(["ssh"] +
236                          get_default_ssh_options(self.ssh_private_key, 120)),
237                 "--delete", "-avz", source, "{}@{}:{}".format(
238                     self.ssh_user, self.ssh_ip, target)
239             ],
240             stdout=redirect or sys.stdout,
241             stderr=redirect or sys.stderr)
242 
243     def rsync_down(self, source, target, redirect=None, check_error=True):
244         self.set_ssh_ip_if_required()
245         self.get_caller(check_error)(
246             [
247                 "rsync", "-e",
248                 " ".join(["ssh"] +
249                          get_default_ssh_options(self.ssh_private_key, 120)),
250                 "-avz", "{}@{}:{}".format(self.ssh_user, self.ssh_ip,
251                                           source), target
252             ],
253             stdout=redirect or sys.stdout,
254             stderr=redirect or sys.stderr)
255 
256     def ssh_cmd(self,
257                 cmd,
258                 connect_timeout=120,
259                 redirect=None,
260                 allocate_tty=False,
261                 emulate_interactive=True,
262                 expect_error=False,
263                 port_forward=None):
264 
265         self.set_ssh_ip_if_required()
266 
267         logger.info("NodeUpdater: Running {} on {}...".format(
268             cmd, self.ssh_ip))
269         ssh = ["ssh"]
270         if allocate_tty:
271             ssh.append("-tt")
272         if emulate_interactive:
273             force_interactive = (
274                 "set -i || true && source ~/.bashrc && "
275                 "export OMP_NUM_THREADS=1 PYTHONWARNINGS=ignore && ")
276             cmd = "bash --login -c {}".format(quote(force_interactive + cmd))
277 
278         if port_forward is None:
279             ssh_opt = []
280         else:
281             ssh_opt = [
282                 "-L", "{}:localhost:{}".format(port_forward, port_forward)
283             ]
284 
285         self.get_caller(expect_error)(
286             ssh + ssh_opt + get_default_ssh_options(self.ssh_private_key,
287                                                     connect_timeout) +
288             ["{}@{}".format(self.ssh_user, self.ssh_ip), cmd],
289             stdout=redirect or sys.stdout,
290             stderr=redirect or sys.stderr)
291 
292 
293 class NodeUpdaterThread(NodeUpdater, Thread):
294     def __init__(self, *args, **kwargs):
295         Thread.__init__(self)
296         NodeUpdater.__init__(self, *args, **kwargs)
297         self.exitcode = -1
298
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/python/ray/autoscaler/updater.py b/python/ray/autoscaler/updater.py
--- a/python/ray/autoscaler/updater.py
+++ b/python/ray/autoscaler/updater.py
@@ -271,9 +271,10 @@
             ssh.append("-tt")
         if emulate_interactive:
             force_interactive = (
-                "set -i || true && source ~/.bashrc && "
+                "true && source ~/.bashrc && "
                 "export OMP_NUM_THREADS=1 PYTHONWARNINGS=ignore && ")
-            cmd = "bash --login -c {}".format(quote(force_interactive + cmd))
+            cmd = "bash --login -c -i {}".format(
+                quote(force_interactive + cmd))
 
         if port_forward is None:
             ssh_opt = []

{"golden_diff": "diff --git a/python/ray/autoscaler/updater.py b/python/ray/autoscaler/updater.py\n--- a/python/ray/autoscaler/updater.py\n+++ b/python/ray/autoscaler/updater.py\n@@ -271,9 +271,10 @@\n             ssh.append(\"-tt\")\n         if emulate_interactive:\n             force_interactive = (\n-                \"set -i || true && source ~/.bashrc && \"\n+                \"true && source ~/.bashrc && \"\n                 \"export OMP_NUM_THREADS=1 PYTHONWARNINGS=ignore && \")\n-            cmd = \"bash --login -c {}\".format(quote(force_interactive + cmd))\n+            cmd = \"bash --login -c -i {}\".format(\n+                quote(force_interactive + cmd))\n \n         if port_forward is None:\n             ssh_opt = []\n", "issue": "[autoscaler] bash 4.4 does not support set -i\n<!--\r\nGeneral questions should be asked on the mailing list ray-dev@googlegroups.com.\r\n\r\nBefore submitting an issue, please fill out the following form.\r\n-->\r\n\r\n### System information\r\n- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**:\r\nUbuntu 17.10\r\n\r\n- **Ray installed from (source or binary)**:\r\nSource\r\n- **Ray version**:\r\ncommit b8811cbe3418ab0d3ea10deaa54947d5bb26cecf\r\n\r\n- **Python version**:\r\n3.6\r\n- **Exact command to reproduce**:\r\nray create_or_update example.yaml\r\n\r\n<!--\r\nYou can obtain the Ray version with\r\n\r\npython -c \"import ray; print(ray.__version__)\"\r\n-->\r\n\r\n### Describe the problem\r\n<!-- Describe the problem clearly here. -->\r\n\r\nAs of bash 4.4, set -i is no longer accepted to create an interactive shell. Consider -t in stead.\r\n\r\n### Source code / logs\r\n<!-- Include any logs or source code that would be helpful to diagnose the problem. If including tracebacks, please include the full traceback. Large logs and files should be attached. Try to provide a reproducible test case that is the bare minimum necessary to generate the problem. -->\r\n\n", "before_files": [{"content": "from __future__ import absolute_import\nfrom __future__ import division\nfrom __future__ import print_function\n\ntry:  # py3\n    from shlex import quote\nexcept ImportError:  # py2\n    from pipes import quote\nimport logging\nimport os\nimport subprocess\nimport sys\nimport time\n\nfrom threading import Thread\n\nfrom ray.autoscaler.tags import TAG_RAY_NODE_STATUS, TAG_RAY_RUNTIME_CONFIG\nfrom ray.autoscaler.log_timer import LogTimer\n\nlogger = logging.getLogger(__name__)\n\n# How long to wait for a node to start, in seconds\nNODE_START_WAIT_S = 300\nSSH_CHECK_INTERVAL = 5\nSSH_CONTROL_PATH = \"/tmp/ray_ssh_sockets\"\n\n\ndef get_default_ssh_options(private_key, connect_timeout):\n    OPTS = [\n        (\"ConnectTimeout\", \"{}s\".format(connect_timeout)),\n        (\"StrictHostKeyChecking\", \"no\"),\n        (\"ControlMaster\", \"auto\"),\n        (\"ControlPath\", \"{}/%C\".format(SSH_CONTROL_PATH)),\n        (\"ControlPersist\", \"yes\"),\n    ]\n\n    return [\"-i\", private_key] + [\n        x for y in ([\"-o\", \"{}={}\".format(k, v)] for k, v in OPTS) for x in y\n    ]\n\n\nclass NodeUpdater(object):\n    \"\"\"A process for syncing files and running init commands on a node.\"\"\"\n\n    def __init__(self,\n                 node_id,\n                 provider_config,\n                 provider,\n                 auth_config,\n                 cluster_name,\n                 file_mounts,\n                 initialization_commands,\n                 setup_commands,\n                 runtime_hash,\n                 process_runner=subprocess,\n                 use_internal_ip=False):\n        self.daemon = True\n        self.process_runner = process_runner\n        self.node_id = node_id\n        self.use_internal_ip = (use_internal_ip or provider_config.get(\n            \"use_internal_ips\", False))\n        self.provider = provider\n        self.ssh_private_key = auth_config[\"ssh_private_key\"]\n        self.ssh_user = auth_config[\"ssh_user\"]\n        self.ssh_ip = None\n        self.file_mounts = {\n            remote: os.path.expanduser(local)\n            for remote, local in file_mounts.items()\n        }\n        self.initialization_commands = initialization_commands\n        self.setup_commands = setup_commands\n        self.runtime_hash = runtime_hash\n\n    def get_caller(self, check_error):\n        if check_error:\n            return self.process_runner.call\n        else:\n            return self.process_runner.check_call\n\n    def get_node_ip(self):\n        if self.use_internal_ip:\n            return self.provider.internal_ip(self.node_id)\n        else:\n            return self.provider.external_ip(self.node_id)\n\n    def wait_for_ip(self, deadline):\n        while time.time() < deadline and \\\n                not self.provider.is_terminated(self.node_id):\n            logger.info(\"NodeUpdater: \"\n                        \"Waiting for IP of {}...\".format(self.node_id))\n            ip = self.get_node_ip()\n            if ip is not None:\n                return ip\n            time.sleep(10)\n\n        return None\n\n    def set_ssh_ip_if_required(self):\n        if self.ssh_ip is not None:\n            return\n\n        # We assume that this never changes.\n        #   I think that's reasonable.\n        deadline = time.time() + NODE_START_WAIT_S\n        with LogTimer(\"NodeUpdater: {}: Got IP\".format(self.node_id)):\n            ip = self.wait_for_ip(deadline)\n            assert ip is not None, \"Unable to find IP of node\"\n\n        self.ssh_ip = ip\n\n        # This should run before any SSH commands and therefore ensure that\n        #   the ControlPath directory exists, allowing SSH to maintain\n        #   persistent sessions later on.\n        with open(\"/dev/null\", \"w\") as redirect:\n            self.get_caller(False)(\n                [\"mkdir\", \"-p\", SSH_CONTROL_PATH],\n                stdout=redirect,\n                stderr=redirect)\n\n            self.get_caller(False)(\n                [\"chmod\", \"0700\", SSH_CONTROL_PATH],\n                stdout=redirect,\n                stderr=redirect)\n\n    def run(self):\n        logger.info(\"NodeUpdater: \"\n                    \"{}: Updating to {}\".format(self.node_id,\n                                                self.runtime_hash))\n        try:\n            m = \"{}: Applied config {}\".format(self.node_id, self.runtime_hash)\n            with LogTimer(\"NodeUpdater: {}\".format(m)):\n                self.do_update()\n        except Exception as e:\n            error_str = str(e)\n            if hasattr(e, \"cmd\"):\n                error_str = \"(Exit Status {}) {}\".format(\n                    e.returncode, \" \".join(e.cmd))\n            logger.error(\"NodeUpdater: \"\n                         \"{}: Error updating {}\".format(\n                             self.node_id, error_str))\n            self.provider.set_node_tags(self.node_id,\n                                        {TAG_RAY_NODE_STATUS: \"update-failed\"})\n            raise e\n\n        self.provider.set_node_tags(\n            self.node_id, {\n                TAG_RAY_NODE_STATUS: \"up-to-date\",\n                TAG_RAY_RUNTIME_CONFIG: self.runtime_hash\n            })\n\n        self.exitcode = 0\n\n    def wait_for_ssh(self, deadline):\n        logger.info(\"NodeUpdater: \"\n                    \"{}: Waiting for SSH...\".format(self.node_id))\n\n        while time.time() < deadline and \\\n                not self.provider.is_terminated(self.node_id):\n            try:\n                logger.debug(\"NodeUpdater: \"\n                             \"{}: Waiting for SSH...\".format(self.node_id))\n                self.ssh_cmd(\n                    \"uptime\",\n                    connect_timeout=5,\n                    redirect=open(\"/dev/null\", \"w\"))\n                return True\n\n            except Exception as e:\n                retry_str = str(e)\n                if hasattr(e, \"cmd\"):\n                    retry_str = \"(Exit Status {}): {}\".format(\n                        e.returncode, \" \".join(e.cmd))\n                logger.debug(\"NodeUpdater: \"\n                             \"{}: SSH not up, retrying: {}\".format(\n                                 self.node_id, retry_str))\n                time.sleep(SSH_CHECK_INTERVAL)\n\n        return False\n\n    def do_update(self):\n        self.provider.set_node_tags(self.node_id,\n                                    {TAG_RAY_NODE_STATUS: \"waiting-for-ssh\"})\n\n        deadline = time.time() + NODE_START_WAIT_S\n        self.set_ssh_ip_if_required()\n\n        # Wait for SSH access\n        with LogTimer(\"NodeUpdater: \" \"{}: Got SSH\".format(self.node_id)):\n            ssh_ok = self.wait_for_ssh(deadline)\n            assert ssh_ok, \"Unable to SSH to node\"\n\n        # Rsync file mounts\n        self.provider.set_node_tags(self.node_id,\n                                    {TAG_RAY_NODE_STATUS: \"syncing-files\"})\n        for remote_path, local_path in self.file_mounts.items():\n            logger.info(\"NodeUpdater: \"\n                        \"{}: Syncing {} to {}...\".format(\n                            self.node_id, local_path, remote_path))\n            assert os.path.exists(local_path), local_path\n            if os.path.isdir(local_path):\n                if not local_path.endswith(\"/\"):\n                    local_path += \"/\"\n                if not remote_path.endswith(\"/\"):\n                    remote_path += \"/\"\n\n            m = \"{}: Synced {} to {}\".format(self.node_id, local_path,\n                                             remote_path)\n            with LogTimer(\"NodeUpdater {}\".format(m)):\n                self.ssh_cmd(\n                    \"mkdir -p {}\".format(os.path.dirname(remote_path)),\n                    redirect=open(\"/dev/null\", \"w\"),\n                )\n                self.rsync_up(\n                    local_path, remote_path, redirect=open(\"/dev/null\", \"w\"))\n\n        # Run init commands\n        self.provider.set_node_tags(self.node_id,\n                                    {TAG_RAY_NODE_STATUS: \"setting-up\"})\n\n        m = \"{}: Initialization commands completed\".format(self.node_id)\n        with LogTimer(\"NodeUpdater: {}\".format(m)):\n            for cmd in self.initialization_commands:\n                self.ssh_cmd(cmd, redirect=open(\"/dev/null\", \"w\"))\n\n        m = \"{}: Setup commands completed\".format(self.node_id)\n        with LogTimer(\"NodeUpdater: {}\".format(m)):\n            for cmd in self.setup_commands:\n                self.ssh_cmd(cmd, redirect=open(\"/dev/null\", \"w\"))\n\n    def rsync_up(self, source, target, redirect=None, check_error=True):\n        self.set_ssh_ip_if_required()\n        self.get_caller(check_error)(\n            [\n                \"rsync\", \"-e\",\n                \" \".join([\"ssh\"] +\n                         get_default_ssh_options(self.ssh_private_key, 120)),\n                \"--delete\", \"-avz\", source, \"{}@{}:{}\".format(\n                    self.ssh_user, self.ssh_ip, target)\n            ],\n            stdout=redirect or sys.stdout,\n            stderr=redirect or sys.stderr)\n\n    def rsync_down(self, source, target, redirect=None, check_error=True):\n        self.set_ssh_ip_if_required()\n        self.get_caller(check_error)(\n            [\n                \"rsync\", \"-e\",\n                \" \".join([\"ssh\"] +\n                         get_default_ssh_options(self.ssh_private_key, 120)),\n                \"-avz\", \"{}@{}:{}\".format(self.ssh_user, self.ssh_ip,\n                                          source), target\n            ],\n            stdout=redirect or sys.stdout,\n            stderr=redirect or sys.stderr)\n\n    def ssh_cmd(self,\n                cmd,\n                connect_timeout=120,\n                redirect=None,\n                allocate_tty=False,\n                emulate_interactive=True,\n                expect_error=False,\n                port_forward=None):\n\n        self.set_ssh_ip_if_required()\n\n        logger.info(\"NodeUpdater: Running {} on {}...\".format(\n            cmd, self.ssh_ip))\n        ssh = [\"ssh\"]\n        if allocate_tty:\n            ssh.append(\"-tt\")\n        if emulate_interactive:\n            force_interactive = (\n                \"set -i || true && source ~/.bashrc && \"\n                \"export OMP_NUM_THREADS=1 PYTHONWARNINGS=ignore && \")\n            cmd = \"bash --login -c {}\".format(quote(force_interactive + cmd))\n\n        if port_forward is None:\n            ssh_opt = []\n        else:\n            ssh_opt = [\n                \"-L\", \"{}:localhost:{}\".format(port_forward, port_forward)\n            ]\n\n        self.get_caller(expect_error)(\n            ssh + ssh_opt + get_default_ssh_options(self.ssh_private_key,\n                                                    connect_timeout) +\n            [\"{}@{}\".format(self.ssh_user, self.ssh_ip), cmd],\n            stdout=redirect or sys.stdout,\n            stderr=redirect or sys.stderr)\n\n\nclass NodeUpdaterThread(NodeUpdater, Thread):\n    def __init__(self, *args, **kwargs):\n        Thread.__init__(self)\n        NodeUpdater.__init__(self, *args, **kwargs)\n        self.exitcode = -1\n", "path": "python/ray/autoscaler/updater.py"}], "after_files": [{"content": "from __future__ import absolute_import\nfrom __future__ import division\nfrom __future__ import print_function\n\ntry:  # py3\n    from shlex import quote\nexcept ImportError:  # py2\n    from pipes import quote\nimport logging\nimport os\nimport subprocess\nimport sys\nimport time\n\nfrom threading import Thread\n\nfrom ray.autoscaler.tags import TAG_RAY_NODE_STATUS, TAG_RAY_RUNTIME_CONFIG\nfrom ray.autoscaler.log_timer import LogTimer\n\nlogger = logging.getLogger(__name__)\n\n# How long to wait for a node to start, in seconds\nNODE_START_WAIT_S = 300\nSSH_CHECK_INTERVAL = 5\nSSH_CONTROL_PATH = \"/tmp/ray_ssh_sockets\"\n\n\ndef get_default_ssh_options(private_key, connect_timeout):\n    OPTS = [\n        (\"ConnectTimeout\", \"{}s\".format(connect_timeout)),\n        (\"StrictHostKeyChecking\", \"no\"),\n        (\"ControlMaster\", \"auto\"),\n        (\"ControlPath\", \"{}/%C\".format(SSH_CONTROL_PATH)),\n        (\"ControlPersist\", \"yes\"),\n    ]\n\n    return [\"-i\", private_key] + [\n        x for y in ([\"-o\", \"{}={}\".format(k, v)] for k, v in OPTS) for x in y\n    ]\n\n\nclass NodeUpdater(object):\n    \"\"\"A process for syncing files and running init commands on a node.\"\"\"\n\n    def __init__(self,\n                 node_id,\n                 provider_config,\n                 provider,\n                 auth_config,\n                 cluster_name,\n                 file_mounts,\n                 initialization_commands,\n                 setup_commands,\n                 runtime_hash,\n                 process_runner=subprocess,\n                 use_internal_ip=False):\n        self.daemon = True\n        self.process_runner = process_runner\n        self.node_id = node_id\n        self.use_internal_ip = (use_internal_ip or provider_config.get(\n            \"use_internal_ips\", False))\n        self.provider = provider\n        self.ssh_private_key = auth_config[\"ssh_private_key\"]\n        self.ssh_user = auth_config[\"ssh_user\"]\n        self.ssh_ip = None\n        self.file_mounts = {\n            remote: os.path.expanduser(local)\n            for remote, local in file_mounts.items()\n        }\n        self.initialization_commands = initialization_commands\n        self.setup_commands = setup_commands\n        self.runtime_hash = runtime_hash\n\n    def get_caller(self, check_error):\n        if check_error:\n            return self.process_runner.call\n        else:\n            return self.process_runner.check_call\n\n    def get_node_ip(self):\n        if self.use_internal_ip:\n            return self.provider.internal_ip(self.node_id)\n        else:\n            return self.provider.external_ip(self.node_id)\n\n    def wait_for_ip(self, deadline):\n        while time.time() < deadline and \\\n                not self.provider.is_terminated(self.node_id):\n            logger.info(\"NodeUpdater: \"\n                        \"Waiting for IP of {}...\".format(self.node_id))\n            ip = self.get_node_ip()\n            if ip is not None:\n                return ip\n            time.sleep(10)\n\n        return None\n\n    def set_ssh_ip_if_required(self):\n        if self.ssh_ip is not None:\n            return\n\n        # We assume that this never changes.\n        #   I think that's reasonable.\n        deadline = time.time() + NODE_START_WAIT_S\n        with LogTimer(\"NodeUpdater: {}: Got IP\".format(self.node_id)):\n            ip = self.wait_for_ip(deadline)\n            assert ip is not None, \"Unable to find IP of node\"\n\n        self.ssh_ip = ip\n\n        # This should run before any SSH commands and therefore ensure that\n        #   the ControlPath directory exists, allowing SSH to maintain\n        #   persistent sessions later on.\n        with open(\"/dev/null\", \"w\") as redirect:\n            self.get_caller(False)(\n                [\"mkdir\", \"-p\", SSH_CONTROL_PATH],\n                stdout=redirect,\n                stderr=redirect)\n\n            self.get_caller(False)(\n                [\"chmod\", \"0700\", SSH_CONTROL_PATH],\n                stdout=redirect,\n                stderr=redirect)\n\n    def run(self):\n        logger.info(\"NodeUpdater: \"\n                    \"{}: Updating to {}\".format(self.node_id,\n                                                self.runtime_hash))\n        try:\n            m = \"{}: Applied config {}\".format(self.node_id, self.runtime_hash)\n            with LogTimer(\"NodeUpdater: {}\".format(m)):\n                self.do_update()\n        except Exception as e:\n            error_str = str(e)\n            if hasattr(e, \"cmd\"):\n                error_str = \"(Exit Status {}) {}\".format(\n                    e.returncode, \" \".join(e.cmd))\n            logger.error(\"NodeUpdater: \"\n                         \"{}: Error updating {}\".format(\n                             self.node_id, error_str))\n            self.provider.set_node_tags(self.node_id,\n                                        {TAG_RAY_NODE_STATUS: \"update-failed\"})\n            raise e\n\n        self.provider.set_node_tags(\n            self.node_id, {\n                TAG_RAY_NODE_STATUS: \"up-to-date\",\n                TAG_RAY_RUNTIME_CONFIG: self.runtime_hash\n            })\n\n        self.exitcode = 0\n\n    def wait_for_ssh(self, deadline):\n        logger.info(\"NodeUpdater: \"\n                    \"{}: Waiting for SSH...\".format(self.node_id))\n\n        while time.time() < deadline and \\\n                not self.provider.is_terminated(self.node_id):\n            try:\n                logger.debug(\"NodeUpdater: \"\n                             \"{}: Waiting for SSH...\".format(self.node_id))\n                self.ssh_cmd(\n                    \"uptime\",\n                    connect_timeout=5,\n                    redirect=open(\"/dev/null\", \"w\"))\n                return True\n\n            except Exception as e:\n                retry_str = str(e)\n                if hasattr(e, \"cmd\"):\n                    retry_str = \"(Exit Status {}): {}\".format(\n                        e.returncode, \" \".join(e.cmd))\n                logger.debug(\"NodeUpdater: \"\n                             \"{}: SSH not up, retrying: {}\".format(\n                                 self.node_id, retry_str))\n                time.sleep(SSH_CHECK_INTERVAL)\n\n        return False\n\n    def do_update(self):\n        self.provider.set_node_tags(self.node_id,\n                                    {TAG_RAY_NODE_STATUS: \"waiting-for-ssh\"})\n\n        deadline = time.time() + NODE_START_WAIT_S\n        self.set_ssh_ip_if_required()\n\n        # Wait for SSH access\n        with LogTimer(\"NodeUpdater: \" \"{}: Got SSH\".format(self.node_id)):\n            ssh_ok = self.wait_for_ssh(deadline)\n            assert ssh_ok, \"Unable to SSH to node\"\n\n        # Rsync file mounts\n        self.provider.set_node_tags(self.node_id,\n                                    {TAG_RAY_NODE_STATUS: \"syncing-files\"})\n        for remote_path, local_path in self.file_mounts.items():\n            logger.info(\"NodeUpdater: \"\n                        \"{}: Syncing {} to {}...\".format(\n                            self.node_id, local_path, remote_path))\n            assert os.path.exists(local_path), local_path\n            if os.path.isdir(local_path):\n                if not local_path.endswith(\"/\"):\n                    local_path += \"/\"\n                if not remote_path.endswith(\"/\"):\n                    remote_path += \"/\"\n\n            m = \"{}: Synced {} to {}\".format(self.node_id, local_path,\n                                             remote_path)\n            with LogTimer(\"NodeUpdater {}\".format(m)):\n                self.ssh_cmd(\n                    \"mkdir -p {}\".format(os.path.dirname(remote_path)),\n                    redirect=open(\"/dev/null\", \"w\"),\n                )\n                self.rsync_up(\n                    local_path, remote_path, redirect=open(\"/dev/null\", \"w\"))\n\n        # Run init commands\n        self.provider.set_node_tags(self.node_id,\n                                    {TAG_RAY_NODE_STATUS: \"setting-up\"})\n\n        m = \"{}: Initialization commands completed\".format(self.node_id)\n        with LogTimer(\"NodeUpdater: {}\".format(m)):\n            for cmd in self.initialization_commands:\n                self.ssh_cmd(cmd, redirect=open(\"/dev/null\", \"w\"))\n\n        m = \"{}: Setup commands completed\".format(self.node_id)\n        with LogTimer(\"NodeUpdater: {}\".format(m)):\n            for cmd in self.setup_commands:\n                self.ssh_cmd(cmd, redirect=open(\"/dev/null\", \"w\"))\n\n    def rsync_up(self, source, target, redirect=None, check_error=True):\n        self.set_ssh_ip_if_required()\n        self.get_caller(check_error)(\n            [\n                \"rsync\", \"-e\",\n                \" \".join([\"ssh\"] +\n                         get_default_ssh_options(self.ssh_private_key, 120)),\n                \"--delete\", \"-avz\", source, \"{}@{}:{}\".format(\n                    self.ssh_user, self.ssh_ip, target)\n            ],\n            stdout=redirect or sys.stdout,\n            stderr=redirect or sys.stderr)\n\n    def rsync_down(self, source, target, redirect=None, check_error=True):\n        self.set_ssh_ip_if_required()\n        self.get_caller(check_error)(\n            [\n                \"rsync\", \"-e\",\n                \" \".join([\"ssh\"] +\n                         get_default_ssh_options(self.ssh_private_key, 120)),\n                \"-avz\", \"{}@{}:{}\".format(self.ssh_user, self.ssh_ip,\n                                          source), target\n            ],\n            stdout=redirect or sys.stdout,\n            stderr=redirect or sys.stderr)\n\n    def ssh_cmd(self,\n                cmd,\n                connect_timeout=120,\n                redirect=None,\n                allocate_tty=False,\n                emulate_interactive=True,\n                expect_error=False,\n                port_forward=None):\n\n        self.set_ssh_ip_if_required()\n\n        logger.info(\"NodeUpdater: Running {} on {}...\".format(\n            cmd, self.ssh_ip))\n        ssh = [\"ssh\"]\n        if allocate_tty:\n            ssh.append(\"-tt\")\n        if emulate_interactive:\n            force_interactive = (\n                \"true && source ~/.bashrc && \"\n                \"export OMP_NUM_THREADS=1 PYTHONWARNINGS=ignore && \")\n            cmd = \"bash --login -c -i {}\".format(\n                quote(force_interactive + cmd))\n\n        if port_forward is None:\n            ssh_opt = []\n        else:\n            ssh_opt = [\n                \"-L\", \"{}:localhost:{}\".format(port_forward, port_forward)\n            ]\n\n        self.get_caller(expect_error)(\n            ssh + ssh_opt + get_default_ssh_options(self.ssh_private_key,\n                                                    connect_timeout) +\n            [\"{}@{}\".format(self.ssh_user, self.ssh_ip), cmd],\n            stdout=redirect or sys.stdout,\n            stderr=redirect or sys.stderr)\n\n\nclass NodeUpdaterThread(NodeUpdater, Thread):\n    def __init__(self, *args, **kwargs):\n        Thread.__init__(self)\n        NodeUpdater.__init__(self, *args, **kwargs)\n        self.exitcode = -1\n", "path": "python/ray/autoscaler/updater.py"}]}