[Storage] Add CloudImplementationFeature for Storage Mounting (#3389)

* prevent single cloud from blocking `sky check`

* add storage_mounting feature

* lint

* lint

* rename

* comment

* lint

Author: romilbhardwaj

sky/clouds/cloud.py

@@ -41,6 +41,7 @@ class CloudImplementationFeatures(enum.Enum):
     SPOT_INSTANCE = 'spot_instance'
     CUSTOM_DISK_TIER = 'custom_disk_tier'
     OPEN_PORTS = 'open_ports'
+    STORAGE_MOUNTING = 'storage_mounting'
 
 
 class Region(collections.namedtuple('Region', ['name'])):

sky/clouds/runpod.py

@@ -38,7 +38,11 @@ class RunPod(clouds.Cloud):
         clouds.CloudImplementationFeatures.DOCKER_IMAGE:
             (f'Docker image is currently not supported on {_REPR}.'),
         clouds.CloudImplementationFeatures.CUSTOM_DISK_TIER:
-            ('Customizing disk tier is not supported yet on RunPod.')
+            ('Customizing disk tier is not supported yet on RunPod.'),
+        clouds.CloudImplementationFeatures.STORAGE_MOUNTING:
+            ('Mounting object stores is not supported on RunPod. To read data '
+             'from object stores on RunPod, use `mode: COPY` to copy the data '
+             'to local disk.'),
     }
     _MAX_CLUSTER_NAME_LEN_LIMIT = 120
     _regions: List[clouds.Region] = []

sky/execution.py

@@ -234,8 +234,8 @@ def _execute(
     # Requested features that some clouds support and others don't.
     requested_features = set()
 
-    if task.num_nodes > 1:
-        requested_features.add(clouds.CloudImplementationFeatures.MULTI_NODE)
+    # Add requested features from the task
+    requested_features |= task.get_required_cloud_features()
 
     backend = backend if backend is not None else backends.CloudVmRayBackend()
     if isinstance(backend, backends.CloudVmRayBackend):

sky/task.py

@@ -1066,6 +1066,30 @@ def add_if_not_none(key, value, no_empty: bool = False):
             })
         return config
 
+    def get_required_cloud_features(
+            self) -> Set[clouds.CloudImplementationFeatures]:
+        """Returns the required features for this task (but not for resources).
+
+        Features required by the resources are checked separately in
+        cloud.get_feasible_launchable_resources().
+
+        INTERNAL: this method is internal-facing.
+        """
+        required_features = set()
+
+        # Multi-node
+        if self.num_nodes > 1:
+            required_features.add(clouds.CloudImplementationFeatures.MULTI_NODE)
+
+        # Storage mounting
+        for _, storage_mount in self.storage_mounts.items():
+            if storage_mount.mode == storage_lib.StorageMode.MOUNT:
+                required_features.add(
+                    clouds.CloudImplementationFeatures.STORAGE_MOUNTING)
+                break
+
+        return required_features
+
     def __rshift__(self, b):
         sky.dag.get_current_dag().add_edge(self, b)