cluster setup script (#111)

README.md

@@ -23,3 +23,42 @@ For a description of our design decisions, see
 3. git clone https://github.com/amplab/ray.git
 4. cd ray
 5. ./setup.sh
+
+## Installing Ray on a cluster
+
+These instructions work on EC2, but they may require some modifications to run
+on your own cluster. In particular, on EC2, running `sudo` does not require a
+password, and we currently don't handle the case where a password is needed.
+
+1. Create a file `nodes.txt` of the IP addresses of the nodes in the cluster.
+For example
+
+        52.50.28.103
+        52.51.210.207
+2. Make sure that the nodes can all communicate with one another. On EC2, this
+can be done by creating a new security group and adding the inbound rule "all
+traffic" and adding the outbound rule "all traffic". Then add all of the nodes
+in your cluster to that security group.
+
+3. Run something like
+    ```
+    python scripts/cluster.py --nodes nodes.txt \
+                              --key-file key.pem \
+                              --username ubuntu \
+                              --installation-directory /home/ubuntu/
+    ```
+where you replace `nodes.txt`, `key.pem`, `ubuntu`, and `/home/ubuntu/` by the
+appropriate values. This assumes that you can connect to each IP address in
+`nodes.txt` with the command
+    ```
+    ssh -i key.pem ubuntu@<ip-address>
+    ```
+4. The previous command should open a Python interpreter. To install Ray on the
+cluster, run `install_ray(node_addresses)` in the interpreter. The interpreter
+should block until the installation has completed.
+5. To check that the installation succeeded, you can ssh to each node, cd into
+the directory `ray/test/`, and run the tests (e.g., `python runtest.py`).
+6. Now that Ray has been installed, you can start the cluster (the scheduler,
+object stores, and workers) with the command `start_ray(node_addresses,
+"/home/ubuntu/ray/test/test_worker.py")`, where the second argument is the path
+on each node in the cluster to the worker code that you would like to use.

scripts/cluster.py

@@ -0,0 +1,83 @@
+# This script can be used to start Ray on an existing cluster.
+#
+# How to use it: Create a file "nodes.txt" that contains a list of the IP
+# addresses of the nodes in the cluster. Put the head node first. This node will
+# host the driver and the scheduler.
+
+import os
+import subprocess
+import socket
+import argparse
+import threading
+import IPython
+
+parser = argparse.ArgumentParser(description="Parse information about the cluster.")
+parser.add_argument("--nodes", type=str, required=True, help="Test file with node IP addresses, one line per address.")
+parser.add_argument("--key-file", type=str, required=True, help="Path to the file that contains the private key.")
+parser.add_argument("--username", type=str, required=True, help="User name for logging in.")
+parser.add_argument("--installation-directory", type=str, required=True, help="The directory in which to install Ray.")
+
+def run_command_over_ssh(node_ip_address, username, key_file, command):
+  full_command = "ssh -i {} {}@{} '{}'".format(key_file, username, node_ip_address, command)
+  subprocess.call([full_command], shell=True)
+  print "Finished running command '{}' on {}@{}.".format(command, username, node_ip_address)
+
+def install_ray_multi_node(node_ip_addresses, username, key_file, installation_directory):
+  def install_ray_over_ssh(node_ip_address, username, key_file, installation_directory):
+    install_ray_command = "sudo apt-get update; sudo apt-get -y install git; mkdir -p {}; cd {}; git clone https://github.com/amplab/ray; cd ray; ./setup.sh".format(installation_directory, installation_directory)
+    run_command_over_ssh(node_ip_address, username, key_file, install_ray_command)
+  threads = []
+  for node_ip_address in node_ip_addresses:
+    t = threading.Thread(target=install_ray_over_ssh, args=(node_ip_address, username, key_file, installation_directory))
+    t.start()
+    threads.append(t)
+  for t in threads:
+    t.join()
+
+def start_ray_multi_node(node_ip_addresses, username, key_file, worker_path, installation_directory):
+  build_directory = os.path.join(installation_directory, "ray/build")
+  start_scheduler_command = "cd {}; nohup ./scheduler {}:10001 > scheduler.out 2> scheduler.err < /dev/null &".format(build_directory, node_ip_addresses[0])
+  run_command_over_ssh(node_ip_addresses[0], username, key_file, start_scheduler_command)
+
+  for i, node_ip_address in enumerate(node_ip_addresses):
+    scripts_directory = os.path.join(installation_directory, "ray/scripts")
+    start_workers_command = "cd {}; python start_workers.py --scheduler-address={}:10001 --node-ip={} --worker-path={} > start_workers.out 2> start_workers.err < /dev/null &".format(scripts_directory, node_ip_addresses[0], node_ip_addresses[i], worker_path)
+    run_command_over_ssh(node_ip_address, username, key_file, start_workers_command)
+
+  print "cluster started; you can start the shell on the head node with:"
+  shell_script_path = os.path.join(args.installation_directory, "ray/scripts/shell.py")
+  print "python {} --scheduler-address={}:10001 --objstore-address={}:20001 --worker-address={}:30001".format(shell_script_path, node_ip_addresses[0], node_ip_addresses[0], node_ip_addresses[0])
+
+def stop_ray_multi_node(node_ip_addresses, username, key):
+  for node_ip_address in node_ip_addresses:
+    kill_cluster_command = "killall scheduler objstore python > /dev/null 2> /dev/null"
+    run_command_over_ssh(node_ip_address, username, key_file, kill_cluster_command)
+
+# Returns true if address is a valid IPv4 address and false otherwise.
+def is_valid_ip(ip_address):
+  try:
+    socket.inet_aton(ip_address)
+    return True
+  except socket.error:
+    return False
+
+if __name__ == "__main__":
+  args = parser.parse_args()
+  username = args.username
+  key_file = args.key_file
+  installation_directory = args.installation_directory
+  node_ip_addresses = map(lambda s: str(s.strip()), open(args.nodes).readlines())
+  for index, node_ip_address in enumerate(node_ip_addresses):
+    if not is_valid_ip(node_ip_address):
+      print "\nWARNING: The string '{}' from line {} in the file {} is not a valid IP address.\n".format(node_ip_address, index + 1, args.nodes)
+
+  def install_ray(node_ip_addresses):
+    install_ray_multi_node(node_ip_addresses, username, key_file, installation_directory)
+
+  def start_ray(node_ip_addresses, worker_path):
+    start_ray_multi_node(node_ip_addresses, username, key_file, worker_path, installation_directory)
+
+  def stop_ray(node_ip_addresses):
+    stop_ray_multi_node(node_ip_addresses, username, key_file)
+
+  IPython.embed()

scripts/shell.py

@@ -0,0 +1,21 @@
+import argparse
+import numpy as np
+
+import ray
+import ray.services as services
+import ray.worker as worker
+
+import ray.arrays.remote as ra
+import ray.arrays.distributed as da
+
+parser = argparse.ArgumentParser(description='Parse addresses for the worker to connect to.')
+parser.add_argument("--scheduler-address", default="127.0.0.1:10001", type=str, help="the scheduler's address")
+parser.add_argument("--objstore-address", default="127.0.0.1:20001", type=str, help="the objstore's address")
+parser.add_argument("--worker-address", default="127.0.0.1:30001", type=str, help="the worker's address")
+
+if __name__ == '__main__':
+  args = parser.parse_args()
+  worker.connect(args.scheduler_address, args.objstore_address, args.worker_address)
+
+  import IPython
+  IPython.embed()

scripts/start_workers.py

@@ -0,0 +1,15 @@
+import argparse
+from ray.services import start_node
+import time
+
+parser = argparse.ArgumentParser(description="Starting workers on a node of the cluster (invoked locally on the node).")
+parser.add_argument("--scheduler-address", type=str, help="Address of the scheduler running on the head node (ip + port).")
+parser.add_argument("--node-ip", type=str, help="IP address of the current worker.")
+parser.add_argument("--num-workers", type=int, default=20, help="Number of workers to be started on the node.")
+parser.add_argument("--worker-path", type=str, help="Path to the worker file.")
+
+if __name__ == "__main__":
+  args = parser.parse_args()
+  start_node(args.scheduler_address, args.node_ip, args.num_workers, worker_path=args.worker_path)
+
+  time.sleep(1000000000) # TODO(pcm): Figure out why object store file handle is closed if we don't do this