Adding VMware platforms support such as vSphere to Ray Autoscaler

doc/source/cluster/vms/getting-started.rst

@@ -51,6 +51,12 @@ Before we start, you will need to install some Python dependencies as follows:
 
             $ pip install -U "ray[default]" google-api-python-client
 
+    .. tab-item:: vSphere
+
+        .. code-block:: shell
+
+            $ pip install vsphere-automation-sdk-python
+
 Next, if you're not set up to use your cloud provider from the command line, you'll have to configure your credentials:
 
 .. tab-set::
@@ -67,6 +73,14 @@ Next, if you're not set up to use your cloud provider from the command line, you
 
         Set the ``GOOGLE_APPLICATION_CREDENTIALS`` environment variable as described in `the GCP docs <https://cloud.google.com/docs/authentication/getting-started>`_.
 
+    .. tab-item:: vSphere
+
+        .. code-block:: shell
+
+            $ export VSPHERE_SERVER = 192.168.0.1 # Enter your vSphere IP
+            $ export VSPHERE_USER = user # Enter your user name
+            $ export VSPHERE_PASSWORD = password # Enter your password
+
 Create a (basic) Python application
 -----------------------------------
 
@@ -200,6 +214,31 @@ A minimal sample cluster configuration file looks as follows:
                 type: gcp
                 region: us-west1
 
+    .. tab-item:: vSphere
+
+        .. code-block:: yaml
+
+            # A unique identifier for the head node and workers of this cluster.
+            cluster_name: minimal
+
+            # Cloud-provider specific configuration.
+            provider:
+                type: vsphere
+
+            auth:
+                ssh_user: ray # The VMs are initialised with an user called ray. 
+
+            available_node_types:
+                ray.head.default:
+                    node_config:
+                        resource_pool: ray # Resource pool where the Ray cluster will get created
+                        library_item: ray-head-debian # OVF file name from which the head will be created
+
+                worker:
+                    node_config:
+                        clone: True # If True, all the workers will be instant-cloned from a frozen VM
+                        library_item: ray-frozen-debian # The OVF file from which a frozen VM will be created
+
 Save this configuration file as ``config.yaml``. You can specify a lot more details in the configuration file: instance types to use, minimum and maximum number of workers to start, autoscaling strategy, files to sync, and more. For a full reference on the available configuration properties, please refer to the :ref:`cluster YAML configuration options reference <cluster-config>`.
 
 After defining our configuration, we will use the Ray cluster launcher to start a cluster on the cloud, creating a designated "head node" and worker nodes. To start the Ray cluster, we will use the :ref:`Ray CLI <ray-cluster-cli>`. Run the following command:

python/ray/autoscaler/_private/command_runner.py

@@ -710,6 +710,11 @@ def run_init(
     ):
         BOOTSTRAP_MOUNTS = ["~/ray_bootstrap_config.yaml", "~/ray_bootstrap_key.pem"]
 
+        # TODO: Maybe add another step after Setup say Post-Setup which will
+        # help in setting up the host after the container comes up.
+        if "VsphereNodeProvider" in str(type(self.ssh_command_runner.provider)):
+            BOOTSTRAP_MOUNTS += ["~/ray_bootstrap_public_key.key"]
+
         specific_image = self.docker_config.get(
             f"{'head' if as_head else 'worker'}_image", self.docker_config.get("image")
         )

python/ray/autoscaler/_private/providers.py

@@ -43,6 +43,12 @@ def _import_azure(provider_config):
     return AzureNodeProvider
 
 
+def _import_vsphere(provider_config):
+    from ray.autoscaler._private.vsphere.node_provider import VsphereNodeProvider
+
+    return VsphereNodeProvider
+
+
 def _import_local(provider_config):
     if "coordinator_address" in provider_config:
         from ray.autoscaler._private.local.coordinator_node_provider import (
@@ -122,6 +128,12 @@ def _load_aws_defaults_config():
     return os.path.join(os.path.dirname(ray_aws.__file__), "defaults.yaml")
 
 
+def _load_vsphere_defaults_config():
+    import ray.autoscaler.vsphere as ray_vsphere
+
+    return os.path.join(os.path.dirname(ray_vsphere.__file__), "defaults.yaml")
+
+
 def _load_gcp_defaults_config():
     import ray.autoscaler.gcp as ray_gcp
 
@@ -152,6 +164,7 @@ def _import_external(provider_config):
     "readonly": _import_readonly,
     "aws": _import_aws,
     "gcp": _import_gcp,
+    "vsphere": _import_vsphere,
     "azure": _import_azure,
     "kubernetes": _import_kubernetes,
     "kuberay": _import_kuberay,
@@ -171,6 +184,7 @@ def _import_external(provider_config):
     "kuberay": "Kuberay",
     "aliyun": "Aliyun",
     "external": "External",
+    "vsphere": "vSphere",
 }
 
 _DEFAULT_CONFIGS = {
@@ -181,6 +195,7 @@ def _import_external(provider_config):
     "azure": _load_azure_defaults_config,
     "aliyun": _load_aliyun_defaults_config,
     "kubernetes": _load_kubernetes_defaults_config,
+    "vsphere": _load_vsphere_defaults_config,
 }
 
 

python/ray/autoscaler/_private/vsphere/config.py

@@ -0,0 +1,224 @@
+import copy
+import logging
+import os
+
+from cryptography.hazmat.backends import default_backend as crypto_default_backend
+from cryptography.hazmat.primitives import serialization as crypto_serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+
+from ray.autoscaler._private.event_system import CreateClusterEvent, global_event_system
+from ray.autoscaler._private.util import check_legacy_fields
+
+PRIVATE_KEY_NAME = "ray-bootstrap-key"
+PRIVATE_KEY_NAME_EXTN = "{}.pem".format(PRIVATE_KEY_NAME)
+
+PUBLIC_KEY_NAME = "ray_bootstrap_public_key"
+PUBLIC_KEY_NAME_EXTN = "{}.key".format(PUBLIC_KEY_NAME)
+
+PRIVATE_KEY_PATH = os.path.expanduser("~/{}.pem".format(PRIVATE_KEY_NAME))
+PUBLIC_KEY_PATH = os.path.expanduser("~/{}.key".format(PUBLIC_KEY_NAME))
+
+USER_DATA_FILE_PATH = os.path.join(os.path.dirname(__file__), "./data/userdata.yaml")
+
+logger = logging.getLogger(__name__)
+
+
+def bootstrap_vsphere(config):
+    # create a copy of the input config to modify
+    config = copy.deepcopy(config)
+
+    add_credentials_into_provider_section(config)
+    # Update library item configs
+    update_vsphere_configs(config)
+
+    # Log warnings if user included deprecated `head_node` or `worker_nodes`
+    # fields. Raise error if no `available_node_types`
+    check_legacy_fields(config)
+
+    # Create new key pair if doesn't exist already
+    create_key_pair()
+
+    # Configure SSH access, using an existing key pair if possible.
+    config = configure_key_pair(config)
+
+    global_event_system.execute_callback(
+        CreateClusterEvent.ssh_keypair_downloaded,
+        {"ssh_key_path": config["auth"]["ssh_private_key"]},
+    )
+
+    return config
+
+
+def add_credentials_into_provider_section(config):
+
+    provider_config = config["provider"]
+
+    # vsphere_config is an optional field as the credentials can also be specified
+    # as env variables so first check verifies if this field is present before
+    # accessing its properties
+    if (
+        "vsphere_config" in provider_config
+        and "credentials" in provider_config["vsphere_config"]
+    ):
+        return
+
+    env_credentials = {}
+    env_credentials["server"] = os.environ["VSPHERE_SERVER"]
+    env_credentials["user"] = os.environ["VSPHERE_USER"]
+    env_credentials["password"] = os.environ["VSPHERE_PASSWORD"]
+
+    provider_config["vsphere_config"] = {}
+    provider_config["vsphere_config"]["credentials"] = env_credentials
+
+
+def update_vsphere_configs(config):
+    """Worker node_config:
+    If clone:False or unspecified:
+        If library_item specified:
+            Create worker from the library_item
+        If library_item unspecified
+            Create worker from head node's library_item
+    If clone:True
+        If library_item unspecified:
+            Terminate
+        If library_item specified:
+            A frozen VM is created from the library item
+            Remaining workers are created from the created frozen VM
+    """
+    available_node_types = config["available_node_types"]
+
+    # Fetch worker: field from the YAML file
+    worker_node = available_node_types["worker"]
+    worker_node_config = worker_node["node_config"]
+
+    # Fetch the head node field name from head_node_type field.
+    head_node_type = config["head_node_type"]
+
+    # Use head_node_type field's value to fetch the head node field
+    head_node = available_node_types[head_node_type]
+    head_node_config = head_node["node_config"]
+
+    # A mandatory constraint enforced by the Ray's YAML validator
+    # is to add resources field for both head and worker nodes.
+    # For example, to specify resources for the worker the
+    # user will specify it in
+    #       worker:
+    #           resources
+    # We copy that resources field into
+    #       worker:
+    #           node_config:
+    #               resources
+    # This enables us to access the field during node creation.
+    # The same happens for head node too.
+    worker_node_config["resources"] = worker_node["resources"]
+    head_node_config["resources"] = head_node["resources"]
+
+    # by default create worker nodes in the head node's resource pool
+    worker_resource_pool = head_node_config["resource_pool"]
+
+    # If different resource pool is provided for worker nodes, use it
+    if "resource_pool" in worker_node_config and worker_node_config["resource_pool"]:
+        worker_resource_pool = worker_node_config["resource_pool"]
+
+    worker_node_config["resource_pool"] = worker_resource_pool
+
+    worker_networks = None
+    worker_datastore = None
+
+    if "networks" in head_node_config and head_node_config["networks"]:
+        worker_networks = head_node_config["networks"]
+
+    if "networks" in worker_node_config and worker_node_config["networks"]:
+        worker_networks = worker_node_config["networks"]
+
+    worker_node_config["networks"] = worker_networks
+
+    if "datastore" in head_node_config and head_node_config["datastore"]:
+        worker_datastore = head_node_config["datastore"]
+
+    if "datastore" in worker_node_config and worker_node_config["datastore"]:
+        worker_datastore = worker_node_config["datastore"]
+
+    worker_node_config["datastore"] = worker_datastore
+
+    if "clone" in worker_node_config and worker_node_config["clone"] == True:
+        if "library_item" not in worker_node_config:
+            raise ValueError(
+                "library_item is mandatory if clone:True is set for worker config"
+            )
+
+        worker_library_item = worker_node_config["library_item"]
+
+        # Create a new object with properties to be used while creating the frozen VM.
+        freeze_vm = {
+            "library_item": worker_library_item,
+            "resource_pool": worker_resource_pool,
+            "resources": worker_node_config["resources"],
+            "networks": worker_networks,
+            "datastore": worker_datastore,
+        }
+
+        # Add the newly created feeze_vm object to head_node config
+        head_node_config["freeze_vm"] = freeze_vm
+
+    elif "clone" not in worker_node_config or worker_node_config["clone"] == False:
+        if "library_item" not in worker_node_config:
+            worker_node_config["library_item"] = head_node_config["library_item"]
+
+
+def create_key_pair():
+
+    # If the files already exists, we don't want to create new keys.
+    # This if condition will currently pass even if there are invalid keys
+    # in those path. TODO: Only return if the keys are valid.
+
+    if os.path.exists(PRIVATE_KEY_PATH) and os.path.exists(PUBLIC_KEY_PATH):
+        logger.info("Key-pair already exist. Not creating new ones")
+        return
+
+    # Generate keys
+    key = rsa.generate_private_key(
+        backend=crypto_default_backend(), public_exponent=65537, key_size=2048
+    )
+
+    private_key = key.private_bytes(
+        crypto_serialization.Encoding.PEM,
+        crypto_serialization.PrivateFormat.PKCS8,
+        crypto_serialization.NoEncryption(),
+    )
+
+    public_key = key.public_key().public_bytes(
+        crypto_serialization.Encoding.OpenSSH, crypto_serialization.PublicFormat.OpenSSH
+    )
+
+    with open(PRIVATE_KEY_PATH, "wb") as content_file:
+        content_file.write(private_key)
+        os.chmod(PRIVATE_KEY_PATH, 0o600)
+
+    with open(PUBLIC_KEY_PATH, "wb") as content_file:
+        content_file.write(public_key)
+
+
+def configure_key_pair(config):
+
+    logger.info("Configure key pairs for copying into the head node.")
+
+    assert os.path.exists(
+        PRIVATE_KEY_PATH
+    ), "Private key file at path {} was not found".format(PRIVATE_KEY_PATH)
+
+    assert os.path.exists(
+        PUBLIC_KEY_PATH
+    ), "Public key file at path {} was not found".format(PUBLIC_KEY_PATH)
+
+    # updater.py file uses the following config to ssh onto the head node
+    # Also, copies the file onto the head node
+    config["auth"]["ssh_private_key"] = PRIVATE_KEY_PATH
+
+    # The path where the public key should be copied onto the remote host
+    public_key_remote_path = "~/{}".format(PUBLIC_KEY_NAME_EXTN)
+
+    # Copy the public key to the remote host
+    config["file_mounts"][public_key_remote_path] = PUBLIC_KEY_PATH
+
+    return config

python/ray/autoscaler/_private/vsphere/data/userdata.yaml

@@ -0,0 +1,10 @@
+#cloud-config
+users:
+- default
+- gecos: Ray
+  groups: sudo, docker
+  lock_passwd: false
+  name: ray
+  passwd: AdminRay
+  primary_group: sudo
+  ssh_authorized_keys: