raising DiskOutOfSpace exception

Author: Alexandra Belousov

.github/workflows/local_tests.yaml

@@ -14,6 +14,11 @@ jobs:
   # DEVELOPMENT INCONSISTENCIES WHEN RUNNING ALL THE LOCAL
   # TESTS TOGETHER.
 
+
+  # PYDEVD_DISABLE_FILE_VALIDATION is an env variable which disables the python debugger on docker cluster. We don't
+  # actually need a python debugger live on docker cluster. Having it causing some unnecessary output in
+  # cluster.run_python()
+
   server-tests-logged-in-level-local:
     runs-on: ubuntu-latest
     steps:
@@ -65,6 +70,7 @@ jobs:
           KITCHEN_TESTER_USERNAME: ${{ secrets.KITCHEN_TESTER_USERNAME }}
           ORG_MEMBER_TOKEN: ${{ secrets.ORG_MEMBER_PROD_TOKEN }}
           ORG_MEMBER_USERNAME: ${{ secrets.ORG_MEMBER_USERNAME }}
+          PYDEVD_DISABLE_FILE_VALIDATION: 1
         run: pytest -v --level local -k "not servertest and not secrettest and not moduletest and not functiontest and not clustertest"
         timeout-minutes: 60
 
@@ -122,6 +128,7 @@ jobs:
           KITCHEN_TESTER_USERNAME: ${{ secrets.KITCHEN_TESTER_USERNAME }}
           ORG_MEMBER_TOKEN: ${{ secrets.ORG_MEMBER_PROD_TOKEN }}
           ORG_MEMBER_USERNAME: ${{ secrets.ORG_MEMBER_USERNAME }}
+          PYDEVD_DISABLE_FILE_VALIDATION: 1
         run: pytest -v --level local -k "moduletest"
         timeout-minutes: 60
 
@@ -178,5 +185,6 @@ jobs:
           KITCHEN_TESTER_USERNAME: ${{ secrets.KITCHEN_TESTER_USERNAME }}
           ORG_MEMBER_TOKEN: ${{ secrets.ORG_MEMBER_PROD_TOKEN }}
           ORG_MEMBER_USERNAME: ${{ secrets.ORG_MEMBER_USERNAME }}
+          PYDEVD_DISABLE_FILE_VALIDATION: 1
         run: pytest -v --level local -k "clustertest"
         timeout-minutes: 60

.github/workflows/local_tests_den_dev.yaml

@@ -12,6 +12,10 @@ jobs:
   # DEVELOPMENT INCONSISTENCIES WHEN RUNNING ALL THE LOCAL
   # TESTS TOGETHER.
 
+  # PYDEVD_DISABLE_FILE_VALIDATION is an env variable which disables the python debugger on docker cluster. We don't
+  # actually need a python debugger live on docker cluster. Having it causing some unnecessary output in
+  # cluster.run_python()
+
   server-tests-logged-in-level-local:
     runs-on: ubuntu-latest
     steps:
@@ -64,6 +68,7 @@ jobs:
           KITCHEN_TESTER_USERNAME: ${{ secrets.KITCHEN_TESTER_USERNAME }}
           ORG_MEMBER_TOKEN: ${{ secrets.ORG_MEMBER_DEV_TOKEN }}
           ORG_MEMBER_USERNAME: ${{ secrets.ORG_MEMBER_USERNAME }}
+          PYDEVD_DISABLE_FILE_VALIDATION: 1
         run: pytest -v --level local -k "not servertest and not secrettest and not moduletest and not functiontest and not clustertest" --api-server-url $API_SERVER_URL
         timeout-minutes: 60
 
@@ -120,6 +125,7 @@ jobs:
           KITCHEN_TESTER_USERNAME: ${{ secrets.KITCHEN_TESTER_USERNAME }}
           ORG_MEMBER_TOKEN: ${{ secrets.ORG_MEMBER_DEV_TOKEN }}
           ORG_MEMBER_USERNAME: ${{ secrets.ORG_MEMBER_USERNAME }}
+          PYDEVD_DISABLE_FILE_VALIDATION: 1
         run: pytest -v --level local -k "moduletest" --api-server-url $API_SERVER_URL
         timeout-minutes: 60
 
@@ -176,5 +182,6 @@ jobs:
           KITCHEN_TESTER_USERNAME: ${{ secrets.KITCHEN_TESTER_USERNAME }}
           ORG_MEMBER_TOKEN: ${{ secrets.ORG_MEMBER_PROD_TOKEN }}
           ORG_MEMBER_USERNAME: ${{ secrets.ORG_MEMBER_USERNAME }}
+          PYDEVD_DISABLE_FILE_VALIDATION: 1
         run: pytest -v --level local -k "clustertest" --api-server-url $API_SERVER_URL
         timeout-minutes: 60

runhouse/constants.py

@@ -115,3 +115,5 @@
 SKY_VENV = "~/skypilot-runtime"
 
 SSH_SKY_SECRET_NAME = "ssh-sky-key"
+
+INSUFFICIENT_DISK_MSG = "No space left on device"

runhouse/exceptions.py

@@ -0,0 +1,27 @@
+# Runhouse exceptions
+
+
+class InsufficientDisk(Exception):
+    """Raised when a process on the cluster fails due to lack of disk space.
+
+    Args:
+        command: The command / process that was run.
+        error_msg: The error message to print.
+    """
+
+    def __init__(
+        self,
+        error_msg: str = None,
+        command: str = None,
+    ) -> None:
+        self.command = command
+        self.error_msg = error_msg
+        msg = (
+            self.error_msg
+            if self.error_msg
+            else f"Command {command} failed"
+            if command
+            else None
+        )
+        msg = f"{msg}. To resolve it, teardown the cluster and re-launch it with larger disk size."
+        super().__init__(msg)

runhouse/resources/hardware/cluster.py

@@ -74,6 +74,8 @@
 from runhouse.resources.hardware.utils import (
     _current_cluster,
     _run_ssh_command,
+    check_disk_sufficiency,
+    get_source_object_size,
     ServerConnectionType,
 )
 from runhouse.resources.resource import Resource
@@ -706,7 +708,7 @@ def _sync_runhouse_to_cluster(
         self._run_setup_step(
             step=ImageSetupStep(
                 step_type=ImageSetupStepType.PACKAGES,
-                reqs=["ray"],
+                reqs=["ray", "psutil"],
                 conda_env_name=conda_env_name,
             ),
             parallel=parallel,
@@ -1486,6 +1488,15 @@ def rsync(
         if not src_node and up and not Path(source).expanduser().exists():
             raise ValueError(f"Could not locate path to sync: {source}.")
 
+        # before syncing the object, making sure there is enough disk space for it. If we don't preform this check in
+        # advance, there might be a case where we start rsyncing the object -> during the rsync no disk space will
+        # be left (because the object is too big) -> the rsync will stack and no results will be return (even failures),
+        # because there is no disk space for running the process nor writing logs.
+        source_size = get_source_object_size(path=source)
+        check_disk_sufficiency(
+            cluster=self, node=node, object_name=source, object_size=source_size
+        )
+
         if up and (node == "all" or (len(self.ips) > 1 and not node)):
             if not parallel:
                 for node in self.ips:
@@ -1942,6 +1953,9 @@ def run_bash_over_ssh(
         if node is None:
             node = self.head_ip
 
+        if conda_env_name:
+            commands = [conda_env_cmd(cmd, conda_env_name) for cmd in commands]
+
         if node == "all":
             res_list = []
             for node in self.ips:
@@ -1955,9 +1969,6 @@ def run_bash_over_ssh(
                 res_list.append(res)
             return res_list
 
-        if conda_env_name:
-            commands = [conda_env_cmd(cmd, conda_env_name) for cmd in commands]
-
         return_codes = self._run_commands_with_runner(
             commands,
             stream_logs=stream_logs,
@@ -2049,7 +2060,8 @@ def _run_commands_with_runner(
         )
 
         for command in commands:
-            logger.info(f"Running command on {self.name}: {command}")
+            if stream_logs:
+                logger.info(f"Running command on {self.name}: {command}")
 
             # set env vars after log statement
             command = f"{env_var_prefix} {command}" if env_var_prefix else command
@@ -2131,12 +2143,7 @@ def run_python(
 
         cmd_prefix = "python3 -c"
         command_str = "; ".join(commands)
-        command_str_repr = (
-            repr(repr(command_str))[2:-2]
-            if self.creds_values.get("password")
-            else command_str
-        )
-        formatted_command = f'{cmd_prefix} "{command_str_repr}"'
+        formatted_command = f'{cmd_prefix} "{command_str}"'
 
         # If invoking a run as part of the python commands also return the Run object
         return_codes = self.run_bash_over_ssh(

runhouse/resources/hardware/sky/command_runner.py

@@ -15,6 +15,7 @@
     log_lib,
     subprocess_utils,
 )
+from runhouse.constants import INSUFFICIENT_DISK_MSG
 
 ##### RH modification #####
 from runhouse.logger import get_logger
@@ -309,8 +310,10 @@ def _rsync(
             time.sleep(backoff.current_backoff())
 
         direction = 'up' if up else 'down'
-        error_msg = (f'Failed to rsync {direction}: {source} -> {target}. '
-                     'Ensure that the network is stable, then retry.')
+        error_msg = f'Failed to rsync {direction}: {source} -> {target}'
+        # means the error is not caused by an insufficient disk space
+        if INSUFFICIENT_DISK_MSG not in stderr:
+            error_msg = error_msg + ' Ensure that the network is stable, then retry.'
         subprocess_utils.handle_returncode(returncode,
                                            command,
                                            error_msg,

runhouse/resources/hardware/sky/subprocess_utils.py

@@ -4,6 +4,9 @@
 
 from runhouse.logger import get_logger
 
+from runhouse.exceptions import InsufficientDisk
+from runhouse.constants import INSUFFICIENT_DISK_MSG
+
 logger = get_logger(__name__)
 
 class CommandError(Exception):
@@ -45,6 +48,8 @@ def handle_returncode(returncode: int,
         if stderr is not None:
             echo(stderr)
 
+        if INSUFFICIENT_DISK_MSG in stderr:
+            raise InsufficientDisk(command=command, error_msg=error_msg)
         if callable(error_msg):
             error_msg = error_msg()
         raise CommandError(returncode, command, error_msg, stderr)

runhouse/resources/hardware/utils.py

@@ -20,6 +20,8 @@
     SKY_VENV,
     TIME_UNITS,
 )
+
+from runhouse.exceptions import InsufficientDisk
 from runhouse.globals import configs, rns_client
 
 from runhouse.logger import get_logger
@@ -258,6 +260,11 @@ def _setup_creds_from_dict(ssh_creds: Dict, cluster_name: str):
     return cluster_secret, ssh_properties
 
 
+###################################
+# Cluster Information Methods
+###################################
+
+
 def is_gpu_cluster(cluster: "Cluster" = None, node: Optional[str] = None):
     if run_setup_command("nvidia-smi", cluster=cluster, node=node)[0] == 0:
         return True
@@ -290,6 +297,46 @@ def detect_cuda_version_or_cpu(cluster: "Cluster" = None, node: Optional[str] =
     return cuda_version
 
 
+def parse_str_to_dict(dict_as_str: str):
+    dict_as_str = dict_as_str.strip().replace("'", '"')
+    return json.loads(dict_as_str)
+
+
+def get_source_object_size(path: str):
+    object_path = Path(path)
+    if object_path.is_dir():
+        # checks the size of all objects in the folder, including subdirectories. rglob is a recursive method, so it
+        # checks all files in all sub-folders, if such exist.
+        return sum(f.stat().st_size for f in object_path.rglob("*") if f.is_file())
+    else:
+        return object_path.stat().st_size
+
+
+def check_disk_sufficiency(
+    cluster: "Cluster", object_size: int, object_name: str, node: str = None
+):
+    # Checking whether the cluster has enough disk space to sync the object.
+    disk_usage = cluster.run_python(
+        commands=["import psutil", "print(psutil.disk_usage('/')._asdict())"],
+        node=node,
+        stream_logs=False,
+        conda_env_name=cluster.conda_env_name,
+    )[0]
+    if node == "all":
+        # If we sync the object to all cluster nodes, we get a list of disk_usage dictionaries, one per node.
+        # Therefore, we need to iterate through each dictionary to ensure that every node has enough disk space.
+        is_disk_sufficient = all(
+            parse_str_to_dict(node_disk_usage[1]).get("free") > object_size
+            for node_disk_usage in disk_usage
+        )
+    else:
+        is_disk_sufficient = parse_str_to_dict(disk_usage[1]).get("free") > object_size
+
+    if not is_disk_sufficient:
+        node = node if node else "head node"
+        raise InsufficientDisk(f"No space left on device ({node}) for {object_name}")
+
+
 def _run_ssh_command(
     address: str,
     ssh_user: str,

runhouse/resources/packages/package.py

@@ -5,6 +5,9 @@
 from pathlib import Path
 from typing import Dict, Optional, Union
 
+from runhouse.constants import INSUFFICIENT_DISK_MSG
+from runhouse.exceptions import InsufficientDisk
+
 from runhouse.globals import obj_store
 
 from runhouse.logger import get_logger
@@ -239,6 +242,19 @@ def _reqs_install_cmd(
         )
         return install_cmd
 
+    def _install_and_validate_output(
+        self, install_cmd: str, cluster: "Cluster" = None, node: Optional[str] = None
+    ):
+        install_res = run_setup_command(install_cmd, cluster=cluster, node=node)
+        retcode = install_res[0]
+        if retcode != 0:
+            stdout, stderr = install_res[1], install_res[2]
+            if INSUFFICIENT_DISK_MSG in stdout or INSUFFICIENT_DISK_MSG in stderr:
+                raise InsufficientDisk(command=install_cmd)
+            raise RuntimeError(
+                f"Pip install {install_cmd} failed, check that the package exists and is available for your platform."
+            )
+
     def _install(
         self,
         cluster: "Cluster" = None,
@@ -287,35 +303,28 @@ def _install(
                 conda_env_name=conda_env_name, cluster=cluster
             )
             logger.info(f"Running via install_method pip: {install_cmd}")
-            retcode = run_setup_command(install_cmd, cluster=cluster, node=node)[0]
-            if retcode != 0:
-                raise RuntimeError(
-                    f"Pip install {install_cmd} failed, check that the package exists and is available for your platform."
-                )
+            self._install_and_validate_output(
+                install_cmd=install_cmd, cluster=cluster, node=node
+            )
 
         elif self.install_method == "conda":
             install_cmd = self._conda_install_cmd(
                 conda_env_name=conda_env_name, cluster=cluster
             )
             logger.info(f"Running via install_method conda: {install_cmd}")
-            retcode = run_setup_command(install_cmd, cluster=cluster, node=node)[0]
-            if retcode != 0:
-                raise RuntimeError(
-                    f"Conda install {install_cmd} failed, check that the package exists and is "
-                    "available for your platform."
-                )
+            self._install_and_validate_output(
+                install_cmd=install_cmd, cluster=cluster, node=node
+            )
 
         elif self.install_method == "reqs":
             install_cmd = self._reqs_install_cmd(
                 conda_env_name=conda_env_name, cluster=cluster
             )
             if install_cmd:
                 logger.info(f"Running via install_method reqs: {install_cmd}")
-                retcode = run_setup_command(install_cmd, cluster=cluster, node=node)[0]
-                if retcode != 0:
-                    raise RuntimeError(
-                        f"Reqs install {install_cmd} failed, check that the package exists and is available for your platform."
-                    )
+                self._install_and_validate_output(
+                    install_cmd=install_cmd, cluster=cluster, node=node
+                )
             else:
                 logger.info(
                     f"{self.install_target.full_local_path_str()}/requirements.txt not found, skipping reqs install"

runhouse/servers/http/http_utils.py

@@ -10,6 +10,8 @@
 
 from pydantic import BaseModel, field_validator
 
+from runhouse.exceptions import InsufficientDisk
+
 from runhouse.logger import get_logger
 
 from runhouse.servers.obj_store import RunhouseStopIteration
@@ -352,6 +354,11 @@ def handle_response(
         ):
             logger.error(f"{system_color}{err_str}: {fn_exception}{reset_color}")
             logger.error(f"{system_color}Traceback: {fn_traceback}{reset_color}")
+
+        # Errno 28 means "No space left on device"
+        if isinstance(fn_exception, OSError) and fn_exception.errno == 28:
+            raise InsufficientDisk()
+
         raise fn_exception
     elif output_type == OutputType.STDOUT:
         res = response_data["data"]