Cluster

Cluster Factory Methods

runhouse.cluster(name: str, host: str | List[str] | None = None, ssh_creds: Dict | str | None = None, server_port: int | None = None, server_host: str | None = None, server_connection_type: ServerConnectionType | str | None = None, launcher: LauncherType | str | None = None, ssl_keyfile: str | None = None, ssl_certfile: str | None = None, domain: str | None = None, den_auth: bool | None = None, image: Image | None = None, load_from_den: bool = True, dryrun: bool = False, **kwargs) → Cluster | OnDemandCluster[source]

Builds an instance of Cluster.

Parameters:

• name (str) – Name for the cluster, to re-use later on.

• host (str or List[str], optional) – Hostname (e.g. domain or name in .ssh/config), IP address, or list of IP addresses for the cluster (the first of which is the head node). (Default: None).

• ssh_creds (dict or str, optional) – SSH credentials, passed as dictionary or the name of an SSHSecret object. Example: ssh_creds={'ssh_user': '...', 'ssh_private_key':'<path_to_key>'} (Default: None).

• server_port (bool, optional) – Port to use for the server. If not provided will use 80 for a server_connection_type of none, 443 for tls and 32300 for all other SSH connection types.

• server_host (bool, optional) – Host from which the server listens for traffic (i.e. the –host argument runhouse server start run on the cluster). Defaults to “0.0.0.0” unless connecting to the server with an SSH connection, in which case localhost is used. (Default: None).

• server_connection_type (ServerConnectionType or str, optional) – Type of connection to use for the Runhouse API server. ssh will use start with server via an SSH tunnel. tls will start the server with HTTPS on port 443 using TLS certs without an SSH tunnel. none will start the server with HTTP without an SSH tunnel. (Default: None).

• launcher (LauncherType or str, optional) – Method for launching the cluster. If set to local, will launch locally via Sky. If set to den, launching will be handled by Runhouse. Currently only relevant for ondemand clusters and kubernetes clusters. (Default: local).

• ssl_keyfile (str, optional) – Path to SSL key file to use for launching the API server with HTTPS. (Default: None).

• ssl_certfile (str, optional) – Path to SSL certificate file to use for launching the API server with HTTPS. (Default: None).

• domain (str, optional) – Domain name for the cluster. Relevant if enabling HTTPs on the cluster. (Default: None).

• den_auth (bool, optional) – Whether to use Den authorization on the server. If True, will validate incoming requests with a Runhouse token provided in the auth headers of the request with the format: {"Authorization": "Bearer <token>"}. (Default: None).

• image (Image, optional) – Default image containing setup steps to run during cluster setup. See Image. (Default: None)

• load_from_den (bool) – Whether to try loading the Cluster resource from Den. (Default: True)

• dryrun (bool) – Whether to create the Cluster if it doesn’t exist, or load a Cluster object as a dryrun. (Default: False)

Returns:

The resulting cluster.

Return type:

Union[Cluster, OnDemandCluster]

Example

>>> # using private key
>>> gpu = rh.cluster(host='<hostname>',
>>>                  ssh_creds={'ssh_user': '...', 'ssh_private_key':'<path_to_key>'},
>>>                  name='rh-a10x').save()

Copied

>>> # using password
>>> gpu = rh.cluster(host='<hostname>',
>>>                  ssh_creds={'ssh_user': '...', 'password':'*****'},
>>>                  name='rh-a10x').save()

Copied

>>> # using the name of an SSHSecret object
>>> gpu = rh.cluster(host='<hostname>',
>>>                  ssh_creds="my_ssh_secret",
>>>                  name='rh-a10x').save()

Copied

>>> # Load cluster from above
>>> reloaded_cluster = rh.cluster(name="rh-a10x")

Copied

runhouse.ondemand_cluster(name: str, instance_type: str | None = None, num_nodes: int | None = None, provider: str | None = None, autostop_mins: int | None = None, use_spot: bool = False, image: Image | None = None, region: str | None = None, memory: int | str | None = None, disk_size: int | str | None = None, num_cpus: int | str | None = None, accelerators: int | str | None = None, open_ports: int | str | List[int] | None = None, sky_kwargs: Dict | None = None, server_port: int | None = None, server_host: int | None = None, server_connection_type: ServerConnectionType | str | None = None, launcher: LauncherType | str | None = None, ssl_keyfile: str | None = None, ssl_certfile: str | None = None, domain: str | None = None, den_auth: bool | None = None, load_from_den: bool = True, dryrun: bool = False, **kwargs) → OnDemandCluster[source]

Builds an instance of OnDemandCluster. Note that region, memory, disk_size, and open_ports are all passed through to SkyPilot’s Resource constructor.

Parameters:

• name (str) – Name for the cluster, to re-use later on.

• instance_type (int, optional) – Type of cloud VM type to use for the cluster, e.g. “r5d.xlarge”. Optional, as may instead choose to specify resource requirements (e.g. memory, disk_size, num_cpus, accelerators).

• num_nodes (int, optional) – Number of nodes to use for the cluster.

• provider (str, optional) – Cloud provider to use for the cluster.

• autostop_mins (int, optional) – Number of minutes to keep the cluster up after inactivity, or -1 to keep cluster up indefinitely. (Default: 60).

• use_spot (bool, optional) – Whether or not to use spot instance.

• region (str, optional) – The region to use for the cluster.

• memory (int or str, optional) – Amount of memory to use for the cluster, e.g. “16” or “16+”.

• disk_size (int or str, optional) – Amount of disk space to use for the cluster, e.g. “100” or “100+”.

• num_cpus (int or str, optional) – Number of CPUs to use for the cluster, e.g. “4” or “4+”.

• accelerators (int or str, optional) – Number of accelerators to use for the cluster, e.g. “A101” or “L4:8”.

• open_ports (int or str or List[int], optional) – Ports to open in the cluster’s security group. Note that you are responsible for ensuring that the applications listening on these ports are secure.

• sky_kwargs (dict, optional) – Additional keyword arguments to pass to the SkyPilot Resource or launch APIs. Should be a dict of the form {“resources”: {<resources_kwargs>}, “launch”: {<launch_kwargs>}}, where resources_kwargs and launch_kwargs will be passed to the SkyPilot Resources API (See SkyPilot docs) and launch API (See SkyPilot docs), respectively. Any arguments which duplicate those passed to the ondemand_cluster factory method will raise an error.

• server_port (bool, optional) – Port to use for the server. If not provided will use 80 for a server_connection_type of none, 443 for tls and 32300 for all other SSH connection types.

• server_host (bool, optional) – Host from which the server listens for traffic (i.e. the –host argument runhouse server start run on the cluster). Defaults to “0.0.0.0” unless connecting to the server with an SSH connection, in which case localhost is used.

• server_connection_type (ServerConnectionType or str, optional) – Type of connection to use for the Runhouse API server. ssh will use start with server via an SSH tunnel. tls will start the server with HTTPS on port 443 using TLS certs without an SSH tunnel. none will start the server with HTTP without an SSH tunnel.

• launcher (LauncherType or str, optional) – Method for launching the cluster. If set to local, will launch locally via Sky. If set to den, launching will be handled by Runhouse. (Default: local).

• ssl_keyfile (str, optional) – Path to SSL key file to use for launching the API server with HTTPS.

• ssl_certfile (str, optional) – Path to SSL certificate file to use for launching the API server with HTTPS.

• domain (str, optional) – Domain name for the cluster. Relevant if enabling HTTPs on the cluster.

• den_auth (bool, optional) – Whether to use Den authorization on the server. If True, will validate incoming requests with a Runhouse token provided in the auth headers of the request with the format: {"Authorization": "Bearer <token>"}. (Default: None).

• image (Image, optional) – Default image containing setup steps to run during cluster setup. See Image. (Default: None)

• load_from_den (bool) – Whether to try loading the Cluster resource from Den. (Default: True)

• dryrun (bool) – Whether to create the Cluster if it doesn’t exist, or load a Cluster object as a dryrun. (Default: False)

Returns:

The resulting cluster.

Return type:

OnDemandCluster

Example

>>> import runhouse as rh
>>> # On-Demand SkyPilot Cluster (OnDemandCluster)
>>> gpu = rh.ondemand_cluster(name='rh-4-a100s',
>>>                  instance_type='A100:4',
>>>                  provider='gcp',
>>>                  autostop_mins=-1,
>>>                  use_spot=True,
>>>                  region='us-east-1',
>>>                  ).save()

Copied

>>> # Load cluster from above
>>> reloaded_cluster = rh.ondemand_cluster(name="rh-4-a100s")

Copied

Cluster Class

class runhouse.Cluster(name: str | None = None, ips: List[str] = None, creds: Secret = None, default_env: Env = None, server_host: str = None, server_port: int = None, ssh_port: int = None, client_port: int = None, server_connection_type: str = None, ssl_keyfile: str = None, ssl_certfile: str = None, domain: str = None, ssh_properties: Dict = None, den_auth: bool = False, dryrun: bool = False, skip_creds: bool = False, image: Image | None = None, **kwargs)[source]

__init__(name: str | None = None, ips: List[str] = None, creds: Secret = None, default_env: Env = None, server_host: str = None, server_port: int = None, ssh_port: int = None, client_port: int = None, server_connection_type: str = None, ssl_keyfile: str = None, ssl_certfile: str = None, domain: str = None, ssh_properties: Dict = None, den_auth: bool = False, dryrun: bool = False, skip_creds: bool = False, image: Image | None = None, **kwargs)[source]

The Runhouse cluster, or system. This is where you can run Functions or access/transfer data between. You can BYO (bring-your-own) cluster by providing cluster IP and ssh_creds, or this can be an on-demand cluster that is spun up/down through SkyPilot, using your cloud credentials.

Note

To build a cluster, please use the factory method cluster().

call(module_name: str, method_name: str, *args, stream_logs: bool = True, run_name: str | None = None, remote: bool = False, run_async: bool = False, save: bool = False, **kwargs)[source]

Call a method on a module that is in the cluster’s object store.

Parameters:

• module_name (str) – Name of the module saved on system.

• method_name (str) – Name of the method.

• stream_logs (bool, optional) – Whether to stream logs from the method call. (Default: True)

• run_name (str, optional) – Name for the run. (Default: None)

• remote (bool, optional) – Return a remote object from the function, rather than the result proper. (Default: False)

• run_async (bool, optional) – Run the method asynchronously and return an awaitable. (Default: False)

• save (bool, optional) – Whether or not to save the call. (Default: False)

• *args – Positional arguments to pass to the method.

• **kwargs – Keyword arguments to pass to the method.

Example

>>> cluster.call("my_module", "my_method", arg1, arg2, kwarg1=kwarg1)

Copied

clear()[source]

Clear the cluster’s object store.

delete(keys: None | str | List[str])[source]

Delete the given items from the cluster’s object store. To delete all items, use cluster.clear()

Parameters:

keys (str or List[str]) – key or list of keys to delete from the object store.

delete_configs(delete_creds: bool = False)[source]

Delete configs for the cluster

disconnect()[source]

Disconnect the RPC tunnel.

Example

>>> cluster.disconnect()

Copied

download_cert()[source]

Download certificate from the cluster (Note: user must have access to the cluster)

enable_den_auth(flush: bool = True)[source]

Enable Den auth on the cluster.

Parameters:

flush (bool, optional) – Whether to flush the auth cache. (Default: True)

endpoint(external: bool = False)[source]

Endpoint for the cluster’s Daemon server.

Parameters:

external (bool, optional) – If True, will only return the external url, and will return None otherwise (e.g. if a tunnel is required). If set to False, will either return the external url if it exists, or will set up the connection (based on connection_type) and return the internal url (including the local connected port rather than the sever port). If cluster is not up, returns None`. (Default: False)

classmethod from_config(config: Dict, dryrun: bool = False, _resolve_children: bool = True)[source]

Load or construct resource from config.

Parameters:

• config (Dict) – Resource config.

• dryrun (bool, optional) – Whether to construct resource or load as dryrun (Default: False)

classmethod from_name(name: str, load_from_den: bool = True, dryrun: bool = False, _alt_options: Dict | None = None, _resolve_children: bool = True)[source]

Load existing Resource via its name.

Parameters:

• name (str) – Name of the resource to load from name.

• load_from_den (bool, optional) – Whether to try loading the module from Den. (Default: True)

• dryrun (bool, optional) – Whether to construct the object or load as dryrun. (Default: False)

get(key: str, default: Any | None = None, remote=False)[source]

Get the result for a given key from the cluster’s object store.

Parameters:

• key (str) – Key to get from the cluster’s object store.

• default (Any, optional) – What to return if the key is not found. To raise an error, pass in KeyError. (Default: None)

• remote (bool, optional) – Whether to get the remote object, rather than the object in full. (Default: False)

install_packages(reqs: List[Package | str], node: str | None = None, conda_env_name: str | None = None)[source]

Install the given packages on the cluster.

Parameters:

• reqs (List[Package or str]) – List of packages to install on cluster and env.

• node (str, optional) – Cluster node to install the package on. If specified, will use ssh to install the package. (Default: None)

• conda_env_name (str, optional) – Name of conda env to install the package in, if relevant. If left empty, defaults to base environment. (Default: None)

Example

>>> cluster.install_packages(reqs=["accelerate", "diffusers"])
>>> cluster.install_packages(reqs=["accelerate", "diffusers"], conda_env_name="my_conda_env")

Copied

is_connected()[source]

Whether the RPC tunnel is up.

Example

>>> connected = cluster.is_connected()

Copied

is_up() → bool[source]

Check if the cluster is up.

Example

>>> rh.cluster("rh-cpu").is_up()

Copied

keys(process: str | None = None)[source]

List all keys in the cluster’s object store.

Parameters:

process (str, optional) – Process in which to list out the keys for.

classmethod list(show_all: bool = False, since: str | None = None, status: str | ClusterStatus | None = None, force: bool = False) → Dict[str, List[Dict]][source]

Loads Runhouse clusters saved in Den and locally via Sky. If filters are provided, only clusters that are matching the filters are returned. If no filters are provided, all running clusters will be returned.

Parameters:

• show_all (bool, optional) – Whether to list all clusters saved in Den. Maximum of 200 will be listed. (Default: False).

• since (str, optional) – Clusters that were active in the specified time period will be returned. Value can be in seconds, minutes, hours or days.

• status (str or ClusterStatus, optional) – Clusters with the provided status will be returned. Options include: running, terminated, initializing, unknown.

• force (bool, optional) – Whether to force a status update for all relevant clusters, or load the latest values. (Default: False).

Examples

>>> Cluster.list(since="75s")
>>> Cluster.list(since="3m")
>>> Cluster.list(since="2h", status="running")
>>> Cluster.list(since="7d")
>>> Cluster.list(show_all=True)

Copied

list_processes()[source]

List all workers on the cluster.

notebook(persist: bool = False, sync_package_on_close: str | None = None, port_forward: int = 8888)[source]

Tunnel into and launch notebook from the cluster.

Example

>>> rh.cluster("test-cluster").notebook()

Copied

on_this_cluster()[source]

Whether this function is being called on the same cluster.

pause_autostop()[source]

Context manager to temporarily pause autostop. Only for OnDemand clusters. There is no autostop for static clusters.

put(key: str, obj: Any, process: str | None = None)[source]

Put the given object on the cluster’s object store at the given key.

Parameters:

• key (str) – Key to assign the object in the object store.

• obj (Any) – Object to put in the object store

• process (str, optional) – Process of the object store to put the object in. (Default: None)

put_resource(resource: Resource, state: Dict | None = None, dryrun: bool = False, process: str | None = None)[source]

Put the given resource on the cluster’s object store. Returns the key (important if name is not set).

Parameters:

• resource (Resource) – Key to assign the object in the object store.

• state (Dict, optional) – Dict of resource attributes to override. (Default: False)

• dryrun (bool, optional) – Whether to put the resource in dryrun mode or not. (Default: False)

• process (str, optional) – Process of the object store to put the object in. (Default: None)

remove_conda_env(conda_env_name: str)[source]

Remove conda env from the cluster.

Parameters:

conda_env_name (str) – Name of conda env to remove from the cluster.

Example

>>> rh.ondemand_cluster("rh-cpu").remove_conda_env("my_conda_env")

Copied

rename(old_key: str, new_key: str)[source]

Rename a key in the cluster’s object store.

Parameters:

• old_key (str) – Original key to rename.

• new_key (str) – Name to reassign the object.

restart_server(_rh_install_url: str | None = None, resync_rh: bool | None = None, restart_ray: bool = True, restart_proxy: bool = False)[source]

Restart the RPC server.

Parameters:

• resync_rh (bool) – Whether to Resync runhouse. If False will not resync Runhouse onto the cluster. If None, will sync if Runhouse is not installed on the cluster or if locally it is installed as editable. (Default: None)

• restart_ray (bool) – Whether to restart Ray. (Default: True)

• restart_proxy (bool) – Whether to restart Caddy on the cluster, if configured. (Default: False)

Example

>>> rh.cluster("rh-cpu").restart_server()

Copied

rsync(source: str, dest: str, up: bool, node: str | None = None, contents: bool = False, filter_options: str | None = None, stream_logs: bool = False, ignore_existing: bool | None = False)[source]

Sync the contents of the source directory into the destination.

Parameters:

• source (str) – The source path.

• dest (str) – The target path.

• up (bool) – The direction of the sync. If True, will rsync from local to cluster. If False will rsync from cluster to local.

• node (Optional[str], optional) – Specific cluster node to rsync to. If not specified will use the address of the cluster’s head node.

• contents (Optional[bool], optional) – Whether the contents of the source directory or the directory itself should be copied to destination. If True the contents of the source directory are copied to the destination, and the source directory itself is not created at the destination. If False the source directory along with its contents are copied ot the destination, creating an additional directory layer at the destination. (Default: False).

• filter_options (Optional[str], optional) – The filter options for rsync.

• stream_logs (Optional[bool], optional) – Whether to stream logs to the stdout/stderr. (Default: False).

• ignore_existing (Optional[bool], optional) – Whether the rsync should skip updating files that already exist on the destination. (Default: False).

Note

Ending source with a slash will copy the contents of the directory into dest, while omitting it will copy the directory itself (adding a directory layer).

run_bash(commands: str | List[str], node: int | str | None = None, process: str | None = None, stream_logs: bool = True, require_outputs: bool = True)[source]

Run bash commands on the cluster through the Runhouse server.

Parameters:

• commands (str or List[str]) – Commands to run on the cluster.

• node (int, str or None) – Node to run the command on. Node can an int referring to the node index, string referring to the ips, or “all” to run on all nodes. If not specified, run the command on the head node. (Default: None)

• process (str or None) – Process to run the command on. (Default: None)

• stream_logs (bool) – Whether to stream logs. (Default: True)

• require_outputs (bool) – Whether to return outputs in addition to status code. (Default: True)

run_bash_over_ssh(commands: str | List[str], node: int | str | None = None, stream_logs: bool = True, require_outputs: bool = True, _ssh_mode: str = 'interactive', conda_env_name: str | None = None)[source]

Run bash commands on the cluster over SSH.

Parameters:

• commands (str or List[str]) – Commands to run on the cluster.

• node (int, str or None) – Node to run the command on. Node can an int referring to the node index, string referring to the ips, or “all” to run on all nodes. If not specified, run the command on the head node. (Default: None)

• stream_logs (bool) – Whether to stream logs. (Default: True)

• require_outputs (bool) – Whether to return outputs in addition to status code. (Default: True)

• conda_env_name (str or None) – Name of conda env to run the command in, if applicable. (Defaut: None)

run_python(commands: List[str], conda_env_name: str | None = None, stream_logs: bool = True, node: str | None = None)[source]

Run a list of python commands on the cluster, or a specific cluster node if its IP is provided.

Parameters:

• commands (List[str]) – List of commands to run.

• process (str, optional) – Process to run the commands in. (Default: None)

• stream_logs (bool, optional) – Whether to stream logs. (Default: True)

• node (str, optional) – Node to run commands on. If not specified, runs on head node. (Default: None)

Example

>>> cpu.run_python(['import numpy', 'print(numpy.__version__)'])
>>> cpu.run_python(["print('hello')"])
>>> cpu.run_python(["print('hello')"], node="3.89.174.234")

Copied

Note

Running Python commands with nested quotes can be finicky. If using nested quotes, try to wrap the outer quote with double quotes (”) and the inner quotes with a single quote (‘).

save(name: str | None = None, overwrite: bool = True, folder: str | None = None)[source]

Overrides the default resource save() method in order to also update the cluster config on the cluster itself.

Parameters:

• name (str, optional) – Name to save the cluster as, if different from its existing name. (Default: None)

• overwrite (bool, optional) – Whether to overwrite the existing saved resource, if it exists. (Default: True)

• folder (str, optional) – Folder to save the config in, if saving locally. If None and saving locally, will be saved in the ~/.rh directory. (Default: None)

property server_address

Address to use in the requests made to the cluster. If creating an SSH tunnel with the cluster, ths will be set to localhost, otherwise will use the cluster’s domain (if provided), or its public IP address.

share(users: str | List[str] | None = None, access_level: ResourceAccess | str = ResourceAccess.READ, visibility: ResourceVisibility | str | None = None, notify_users: bool = True, headers: Dict | None = None) → Tuple[Dict[str, ResourceAccess], Dict[str, ResourceAccess]][source]

Grant access to the resource for a list of users (or a single user). By default, the user will receive an email notification of access (if they have a Runhouse account) or instructions on creating an account to access the resource. If visibility is set to public, users will not be notified.

Note

You can only grant access to other users if you have write access to the resource.

Parameters:

• users (Union[str, list], optional) – Single user or list of user emails and / or runhouse account usernames. If none are provided and visibility is set to public, resource will be made publicly available to all users. (Default: None)

• access_level (ResourceAccess, optional) – Access level to provide for the resource. (Default: read).

• visibility (ResourceVisibility, optional) – Type of visibility to provide for the shared resource. By default, the visibility is private. (Default: None)

• notify_users (bool, optional) – Whether to send an email notification to users who have been given access. Note: This is relevant for resources which are not shareable. (Default: True)

• headers (Dict, optional) – Request headers to provide for the request to Den. Contains the user’s auth token. Example: {"Authorization": f"Bearer {token}"}

Returns:

added_users:

Users who already have a Runhouse account and have been granted access to the resource.

new_users:

Users who do not have Runhouse accounts and received notifications via their emails.

valid_users:

Set of valid usernames and emails from users parameter.

Return type:

Tuple(Dict, Dict, Set)

Example

>>> # Write access to the resource for these specific users.
>>> # Visibility will be set to private (users can search for and view resource in Den dashboard)
>>> my_resource.share(users=["username1", "user2@gmail.com"], access_level='write')

Copied

>>> # Make resource public, with read access to the resource for all users
>>> my_resource.share(visibility='public')

Copied

ssh()[source]

SSH into the cluster

Example

>>> rh.cluster("rh-cpu").ssh()

Copied

start_server(_rh_install_url: str | None = None, resync_rh: bool | None = None, restart_ray: bool = True, restart_proxy: bool = False)[source]

Restart the RPC server.

Parameters:

• resync_rh (bool) – Whether to Resync runhouse. If False will not resync Runhouse onto the cluster. If None, will sync if Runhouse is not installed on the cluster or if locally it is installed as editable. (Default: None)

• restart_ray (bool) – Whether to restart Ray. (Default: True)

• restart_proxy (bool) – Whether to restart Caddy on the cluster, if configured. (Default: False)

Example

>>> rh.cluster("rh-cpu").start_server()

Copied

status(send_to_den: bool = False)[source]

Load the status of the Runhouse daemon running on a cluster.

Parameters:

send_to_den (bool, optional) – Whether to send and update the status in Den. Only applies to clusters that are saved to Den. (Default: False)

stop_server(stop_ray: bool = False, cleanup_actors: bool = True, conda_env_name: str | None = None)[source]

Stop the RPC server.

Parameters:

• stop_ray (bool, optional) – Whether to stop Ray. (Default: True)

• process (str, optional) – Specified process to stop the server on. (Default: None)

• cleanup_actors (bool, optional) – Whether to kill all Ray actors. (Default: True)

sync_secrets(providers: List[str] | None = None, process: str | None = None)[source]

Send secrets for the given providers.

Parameters:

• providers (List[str] or None, optional) – List of providers to send secrets for. If None, all providers configured in the environment will by sent. (Default: None)

• process (str, optional) – Process to sync secrets into, if setting env vars. (Default: None)

Example

>>> cpu.sync_secrets(secrets=["aws", "lambda"])

Copied

up_if_not(verbose: bool = True)[source]

Bring up the cluster if it is not up. No-op if cluster is already up. This only applies to on-demand clusters, and has no effect on self-managed clusters.

Parameters:

verbose (bool, optional) – Whether to stream logs from Den if the cluster is being launched. Only relevant if launching via Den. (Default: True)

Example

>>> rh.cluster("rh-cpu").up_if_not()

Copied

OnDemandCluster Class

A OnDemandCluster is a cluster that uses SkyPilot functionality underneath to handle various cluster properties.

class runhouse.OnDemandCluster(name, instance_type: str = None, num_nodes: int = None, provider: str = None, default_env: Env = None, dryrun: bool = False, autostop_mins: int = None, use_spot: bool = False, memory: int | str = None, disk_size: int | str = None, num_cpus: int | str = None, accelerators: str = None, open_ports: int | str | List[int] = None, server_host: int = None, server_port: int = None, server_connection_type: str = None, launcher: str = None, ssl_keyfile: str = None, ssl_certfile: str = None, domain: str = None, den_auth: bool = False, region: str = None, sky_kwargs: Dict = None, **kwargs)[source]

__init__(name, instance_type: str = None, num_nodes: int = None, provider: str = None, default_env: Env = None, dryrun: bool = False, autostop_mins: int = None, use_spot: bool = False, memory: int | str = None, disk_size: int | str = None, num_cpus: int | str = None, accelerators: str = None, open_ports: int | str | List[int] = None, server_host: int = None, server_port: int = None, server_connection_type: str = None, launcher: str = None, ssl_keyfile: str = None, ssl_certfile: str = None, domain: str = None, den_auth: bool = False, region: str = None, sky_kwargs: Dict = None, **kwargs)[source]

On-demand SkyPilot Cluster.

Note

To build a cluster, please use the factory method cluster().

async a_up(capture_output: bool | str = True)[source]

Up the cluster async in another process, so it can be parallelized and logs can be captured sanely.

capture_output: If True, supress the output of the cluster creation process. If False, print the output normally. If a string, write the output to the file at that path.

accelerators()[source]

Returns the acclerator type, or None if is a CPU.

static cluster_ssh_key(path_to_file: Path)[source]

Retrieve SSH key for the cluster.

Parameters:

path_to_file (Path) – Path of the private key associated with the cluster.

Example

>>> ssh_priv_key = rh.ondemand_cluster("rh-cpu").cluster_ssh_key("~/.ssh/id_rsa")

Copied

endpoint(external: bool = False)[source]

Endpoint for the cluster’s Daemon server.

Parameters:

external (bool, optional) – If True, will only return the external url, and will return None otherwise (e.g. if a tunnel is required). If set to False, will either return the external url if it exists, or will set up the connection (based on connection_type) and return the internal url (including the local connected port rather than the sever port). If cluster is not up, returns None`. (Default: False)

get_instance_type()[source]

Returns instance type of the cluster.

is_up() → bool[source]

Whether the cluster is up.

Example

>>> rh.ondemand_cluster("rh-cpu").is_up()

Copied

keep_warm(mins: int = -1)[source]

Keep the cluster warm for given number of minutes after inactivity.

Parameters:

mins (int) – Amount of time (in min) to keep the cluster warm after inactivity. If set to -1, keep cluster warm indefinitely. (Default: -1)

num_cpus()[source]

Return the number of CPUs for a CPU cluster.

pause_autostop()[source]

Context manager to temporarily pause autostop.

Example

>>> with rh.ondemand_cluster.pause_autostop():
>>>     rh.ondemand_cluster.run(["python train.py"])

Copied

ssh(node: str | None = None)[source]

SSH into the cluster.

Parameters:

node – Node to SSH into. If no node is specified, will SSH onto the head node. (Default: None)

Example

>>> rh.ondemand_cluster("rh-cpu").ssh()
>>> rh.ondemand_cluster("rh-cpu", node="3.89.174.234").ssh()

Copied

teardown(verbose: bool = True)[source]

Teardown cluster.

Parameters:

verbose (bool, optional) – Whether to stream logs from Den when the cluster is being downed. Only relevant when tearing down via Den. (Default: True)

Example

>>> rh.ondemand_cluster("rh-cpu").teardown()

Copied

teardown_and_delete(verbose: bool = True)[source]

Teardown cluster and delete it from configs.

Parameters:

verbose (bool, optional) – Whether to stream logs from Den when the cluster is being downed. Only relevant when tearing down via Den. (Default: True)

Example

>>> rh.ondemand_cluster("rh-cpu").teardown_and_delete()

Copied

up(verbose: bool = True, force: bool = False)[source]

Up the cluster.

Parameters:

• verbose (bool, optional) – Whether to stream logs from Den when the cluster is being launched. Only relevant if launching via Den. (Default: True)

• force (bool, optional) – Whether to launch the cluster even if one with the same configs already exists. Only relevant if launching via Den. (Default: False)

Example

>>> rh.ondemand_cluster("rh-cpu").up()

Copied

$ sky check

<cloud-provider>:
  vpc: <vpc-name>

Cluster Authentication & Verification

Runhouse provides a couple of options to manage the connection to the Runhouse API server running on a cluster.

import runhouse as rh

def concat(a: str, b: str):
    return a + b

# Launch a cluster with TLS and Den Auth enabled
cpu = rh.ondemand_cluster(instance_type="m5.xlarge",
                          provider="aws",
                          name="rh-cluster",
                          den_auth=True,
                          open_ports=[443],
                          server_connection_type="tls").up_if_not()

# Remote function stub which lives on the cluster
remote_func = rh.function(concat).to(cpu)

# Save to Runhouse Den
remote_func.save()

# Give read access to the function to another user - this will allow them to call this service remotely
# and view the function metadata in Runhouse Den
remote_func.share("user1@gmail.com", access_level="read")

# This other user (user1) can then call the function remotely from any python environment
res = remote_func("run", "house")
>> print(res)
>> "runhouse"

$ curl -X GET "https://<DOMAIN>/concat/call?a=run&b=house"
-H "Content-Type: application/json" -H "Authorization: Bearer <CLUSTER-TOKEN>"

Using a Custom Domain

Runhouse supports using custom domains for deploying your apps and services. You can provide the domain ahead of time before launching the cluster by specifying the domain argument:

cluster = rh.cluster(name="rh-serving-cpu",
                     domain="<your domain>",
                     instance_type="m5.xlarge",
                     server_connection_type="tls",
                     open_ports=[443]).up_if_not()

If you have an existing cluster, you can also configure a domain by including the IP and domain when initializing the Runhouse cluster object:

cluster = rh.cluster(name="rh-serving-cpu",
                     ips=["<public IP>"],
                     domain="<your domain>",
                     server_connection_type="tls",
                     open_ports=[443]).up_if_not()

Now we can send modules or functions to our cluster and seamlessly create endpoints which we can then share and call from anywhere.

Let’s take a look at an example of how to deploy a simple LangChain RAG app.

Once the app has been created and sent to the cluster, we can call it via HTTP directly:

import requests

resp = requests.get("https://<domain>/basic_rag_app/invoke?user_prompt=<prompt>")
print(resp.json())

Or via cURL:

$ curl "https://<domain>/basic_rag_app/invoke?user_prompt=<prompt>"