Repository: skypilot-org/skypilot
Pull request #2326 (2023-07-30 11:22:31)

[Storage/R2] Fix possible invalid bucket name.
During the work on adding minio support I have encountered an issue related to Bucket name where I fixed it in:
https://github.com/skypilot-org/skypilot/pull/2315/commits/e4424db1ba12bf99f0ff8fbeeaab1dc8445b7523

I then noticed that this possible issue can also exist in R2 (which is also S3 compatible). I added this fix in this PR.

I can not validate it since I do not have an R2 account.

- [ v ] Code formatting: `bash format.sh`


aviweit: Hi @romilbhardwaj , please feel free to accept the PR if it helps.
Thanks.
romilbhardwaj: Thanks @aviweit! Looping in @landscapepainter to take a look.
landscapepainter: There's an error currently in our master branch failing for `test_optimizer_random_dag.py`https://github.com/skypilot-org/skypilot/issues/2320. Let's mege with master after the fix. Then, should be ready to go after the pass.

Patch:
diff --git a/docs/source/getting-started/installation.rst b/docs/source/getting-started/installation.rst
index 2fc231108..714312221 100644
--- a/docs/source/getting-started/installation.rst
+++ b/docs/source/getting-started/installation.rst
@@ -127,7 +127,7 @@ Lambda Cloud
 IBM
 ~~~~~~~~~
 
-To access IBM's services, store the following fields in ``~/.ibm/credentials.yaml``:
+To access IBM's VPC service, store the following fields in ``~/.ibm/credentials.yaml``:
 
 .. code-block:: text
 
@@ -141,6 +141,23 @@ To access IBM's services, store the following fields in ``~/.ibm/credentials.yam
   Stock images aren't currently providing ML tools out of the box.
   Create private images with the necessary tools (e.g. CUDA), by following the IBM segment in `this documentation <https://github.com/skypilot-org/skypilot/blob/master/docs/source/reference/yaml-spec.rst>`_.
 
+To access IBM's Cloud Object Storage (COS), append the following fields to the credentials file:
+
+.. code-block:: text
+
+  access_key_id: <access_key_id>
+  secret_access_key: <secret_key_id>
+
+To get :code:`access_key_id` and :code:`secret_access_key` use the IBM web console:
+
+1. Create/Select a COS instance from the `web console <https://cloud.ibm.com/objectstorage/>`__.
+2. From "Service Credentials" tab, click "New Credential" and toggle "Include HMAC Credential".
+3. Copy "secret_access_key" and "access_key_id" to file.
+
+Finally, install `rclone <https://rclone.org/>`_ via: ``curl https://rclone.org/install.sh | sudo bash``
+
+.. note::
+  :code:`sky check` does not reflect IBM COS's enabled status. :code:`IBM: enabled` only guarantees that IBM VM instances are enabled.
 
 Oracle Cloud Infrastructure (OCI)
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
diff --git a/docs/source/reference/storage.rst b/docs/source/reference/storage.rst
index b8ee43939..dcf9958b5 100644
--- a/docs/source/reference/storage.rst
+++ b/docs/source/reference/storage.rst
@@ -8,7 +8,7 @@ files required by the task. Think of it as a bucket of files that can be attache
 to your task. Compared to file_mounts, storage is faster and
 can persist across runs, requiring fewer uploads from your local machine.
 Behind the scenes, storage automatically uploads all data in the source
-to a backing object store in a particular cloud (S3/GCS/R2).
+to a backing object store in a particular cloud (S3/GCS/R2/IBM).
 
 A storage object is used by "mounting" it to a task. On mounting, the data
 specified in the source becomes available at the destination mount path.
@@ -32,7 +32,7 @@ SkyPilot Storage can be used by specifying additional fields in the
 :code:`file_mounts`. By default, :code:`file_mounts` uses rsync to
 directly copy files from local to remote VM.
 However, you can have them backed by SkyPilot Storage, which uploads
-the files to a cloud store (e.g. S3, GCS, R2) and have them persist there by
+the files to a cloud store (e.g. S3, GCS, R2, or IBM) and have them persist there by
 specifying the :code:`name`, :code:`source` and :code:`persistent` fields. By
 enabling persistence, file_mount sync can be made significantly faster.
 
@@ -99,6 +99,12 @@ and storage mounting:
       # storage to local disk.
       /train-00001-of-01024: gs://cloud-tpu-test-datasets/fake_imagenet/train-00001-of-01024
 
+      # *** Copying files from IBM COS ***
+      #
+      # This re-uses a predefined bucket and copies its contents directly to /datasets-cos. 
+      # Users must provide the region their bucket resides in, e.g. cos://us-east/bucket-name.
+      /datasets-cos: cos://<region-of-bucket>/<bucket-name>
+
       # *** Persistent Data Storage by copying from S3 ***
       #
       # This uses SkyPilot Storage to first create a S3 bucket named sky-dataset,
@@ -110,7 +116,7 @@ and storage mounting:
       /datasets-storage:
         name: sky-dataset-romil # Make sure this name is unique or you own this bucket
         source: ~/datasets
-        store: s3 # Could be either of [s3, gcs, r2]; default: None
+        store: s3 # Could be either of [s3, gcs, r2, ibm]; default: None
         persistent: True  # Defaults to True, can be set to false.
         mode: COPY  # Defaults to MOUNT if not specified
 
@@ -272,12 +278,12 @@ Storage YAML reference
       sky.Storage.source: str
         The source attribute specifies the local path that must be made available
         in the storage object. It can either be a local path or a list of local
-        paths or it can be a remote path (s3://, gs://, r2://).
+        paths or it can be a remote path (s3://, gs://, r2://, cos://<region_name>).
         If the source is local, data is uploaded to the cloud to an appropriate
-        object store (s3, gcs or r2). If the path is remote, the data is copied
+        object store (s3, gcs, r2, or ibm). If the path is remote, the data is copied
         or mounted directly (see mode flag below).
 
-      sky.Storage.store: str; either of 's3', 'gcs' or 'r2'
+      sky.Storage.store: str; either of 's3', 'gcs', 'r2', 'ibm'
         If you wish to force sky.Storage to be backed by a specific cloud object
         store, you can specify it here. If not specified, SkyPilot chooses the
         appropriate object store based on the source path and task's cloud provider.
diff --git a/sky/adaptors/ibm.py b/sky/adaptors/ibm.py
index 0395a4820..9cffdca89 100644
--- a/sky/adaptors/ibm.py
+++ b/sky/adaptors/ibm.py
@@ -8,6 +8,7 @@ import os
 import json
 import requests
 import functools
+import multiprocessing
 
 CREDENTIAL_FILE = '~/.ibm/credentials.yaml'
 logger = sky_logging.init_logger(__name__)
@@ -15,6 +16,8 @@ logger = sky_logging.init_logger(__name__)
 ibm_vpc = None
 ibm_cloud_sdk_core = None
 ibm_platform_services = None
+ibm_boto3 = None
+ibm_botocore = None
 
 
 def import_package(func):
@@ -22,14 +25,19 @@ def import_package(func):
     @functools.wraps(func)
     def wrapper(*args, **kwargs):
         global ibm_vpc, ibm_cloud_sdk_core, ibm_platform_services
+        global ibm_boto3, ibm_botocore
         if None in [ibm_vpc, ibm_cloud_sdk_core, ibm_platform_services]:
             try:
                 import ibm_vpc as _ibm_vpc
                 import ibm_cloud_sdk_core as _ibm_cloud_sdk_core
                 import ibm_platform_services as _ibm_platform_services
+                import ibm_boto3 as _ibm_boto3
+                import ibm_botocore as _ibm_botocore
                 ibm_vpc = _ibm_vpc
                 ibm_cloud_sdk_core = _ibm_cloud_sdk_core
                 ibm_platform_services = _ibm_platform_services
+                ibm_boto3 = _ibm_boto3
+                ibm_botocore = _ibm_botocore
             except ImportError:
                 raise ImportError(
                     'Failed to import dependencies for IBM. '
@@ -40,14 +48,23 @@ def import_package(func):
 
 
 def read_credential_file():
-    with open(os.path.expanduser(CREDENTIAL_FILE), encoding='utf-8') as f:
-        return yaml.safe_load(f)
+    try:
+        with open(os.path.expanduser(CREDENTIAL_FILE), 'r',
+                  encoding='utf-8') as f:
+            return yaml.safe_load(f)
+    except FileNotFoundError:
+        return False
 
 
 def get_api_key():
     return read_credential_file()['iam_api_key']
 
 
+def get_hmac_keys():
+    cred_file = read_credential_file()
+    return cred_file['access_key_id'], cred_file['secret_access_key']
+
+
 @import_package
 def _get_authenticator():
     return ibm_cloud_sdk_core.authenticators.IAMAuthenticator(get_api_key())
@@ -105,3 +122,63 @@ def search_client():
 def tagging_client():
     return ibm_platform_services.GlobalTaggingV1(
         authenticator=_get_authenticator())
+
+
+@import_package
+def get_cos_client(region: str = 'us-east'):
+    """Returns an IBM COS client object.
+      Using process lock to protect not multi process
+      safe Boto3.session, which is invoked (default session) by
+      boto3.client.
+
+    Args:
+        region (str, optional): Client endpoint. Defaults to 'us-east'.
+
+    Returns:
+        IBM COS client
+    """
+    access_key_id, secret_access_key = get_hmac_keys()
+    with _get_global_process_lock():
+        return ibm_boto3.client(  # type: ignore[union-attr]
+            service_name='s3',
+            aws_access_key_id=access_key_id,
+            aws_secret_access_key=secret_access_key,
+            endpoint_url=
+            f'https://s3.{region}.cloud-object-storage.appdomain.cloud')
+
+
+@import_package
+def get_cos_resource(region: str = 'us-east'):
+    """Returns an IBM COS Resource object.
+      Using process lock to protect not multi process safe
+      boto3.Resource.
+
+    Args:
+        region (str, optional): Resource Endpoint. Defaults to 'us-east'.
+
+    Returns:
+        IBM COS Resource
+    """
+    access_key_id, secret_access_key = get_hmac_keys()
+    with _get_global_process_lock():
+        return ibm_boto3.resource(  # type: ignore[union-attr]
+            's3',
+            aws_access_key_id=access_key_id,
+            aws_secret_access_key=secret_access_key,
+            endpoint_url=
+            f'https://s3.{region}.cloud-object-storage.appdomain.cloud')
+
+
+def _get_global_process_lock():
+    """returns a multi-process lock.
+
+    lazily initializes a global multi-process lock if it wasn't
+      already initialized.
+    Necessary when process are spawned without a shared lock.
+    """
+    global global_process_lock  # pylint: disable=global-variable-undefined
+
+    if 'global_process_lock' not in globals():
+        global_process_lock = multiprocessing.Lock()
+
+    return global_process_lock
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index 502dec408..5303a5382 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -3480,7 +3480,7 @@ class CloudVmRayBackend(backends.Backend['CloudVmRayResourceHandle']):
                 returncode = -1
 
             if vpc_found:
-                # # pylint: disable=line-too-long E1136
+                # pylint: disable=line-too-long E1136
                 # Delete VPC and it's associated resources
                 vpc_provider = IBMVPCProvider(
                     config_provider['resource_group_id'], region, cluster_name)
diff --git a/sky/cli.py b/sky/cli.py
index 7fa66175b..380a5eca0 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -2332,9 +2332,15 @@ def start(
     to_start = []
 
     if not clusters and not all:
-        raise click.UsageError(
-            'sky start requires either a cluster name (see `sky status`) '
-            'or --all.')
+        # UX: frequently users may have only 1 cluster. In this case, be smart
+        # and default to that unique choice.
+        all_cluster_names = global_user_state.get_cluster_names_start_with('')
+        if len(all_cluster_names) <= 1:
+            clusters = all_cluster_names
+        else:
+            raise click.UsageError(
+                '`sky start` requires either a cluster name or glob '
+                '(see `sky status`), or the -a/--all flag.')
 
     if all:
         if len(clusters) > 0:
@@ -2348,7 +2354,11 @@ def start(
             if cluster['name'] not in backend_utils.SKY_RESERVED_CLUSTER_NAMES
         ]
 
-    if clusters:
+    if not clusters:
+        click.echo('Cluster(s) not found (tip: see `sky status`). Do you '
+                   'mean to use `sky launch` to provision a new cluster?')
+        return
+    else:
         # Get GLOB cluster names
         clusters = _get_glob_clusters(clusters)
         local_clusters = onprem_utils.check_and_get_local_clusters()
@@ -2603,9 +2613,16 @@ def _down_or_stop_clusters(
     else:
         command = 'stop'
     if not names and apply_to_all is None:
-        raise click.UsageError(
-            f'`sky {command}` requires either a cluster name (see `sky status`)'
-            ' or --all.')
+        # UX: frequently users may have only 1 cluster. In this case, 'sky
+        # stop/down' without args should be smart and default to that unique
+        # choice.
+        all_cluster_names = global_user_state.get_cluster_names_start_with('')
+        if len(all_cluster_names) <= 1:
+            names = all_cluster_names
+        else:
+            raise click.UsageError(
+                f'`sky {command}` requires either a cluster name or glob '
+                '(see `sky status`), or the -a/--all flag.')
 
     operation = 'Terminating' if down else 'Stopping'
     if idle_minutes_to_autostop is not None:
@@ -2695,7 +2712,7 @@ def _down_or_stop_clusters(
     usage_lib.record_cluster_name_for_current_operation(clusters)
 
     if not clusters:
-        click.echo('\nCluster(s) not found (tip: see `sky status`).')
+        click.echo('Cluster(s) not found (tip: see `sky status`).')
         return
 
     if not no_confirm and len(clusters) > 0:
diff --git a/sky/cloud_stores.py b/sky/cloud_stores.py
index f88d3c136..85a0f80ff 100644
--- a/sky/cloud_stores.py
+++ b/sky/cloud_stores.py
@@ -12,7 +12,8 @@ import urllib.parse
 
 from sky.clouds import gcp
 from sky.data import data_utils
-from sky.adaptors import aws, cloudflare
+from sky.adaptors import aws, cloudflare, ibm
+from sky.data.data_utils import Rclone
 
 
 class CloudStorage:
@@ -202,9 +203,11 @@ class R2CloudStorage(CloudStorage):
     def make_sync_file_command(self, source: str, destination: str) -> str:
         """Downloads a file using AWS CLI."""
         endpoint_url = cloudflare.create_endpoint()
+        if 'r2://' in source:
+            source = source.replace('r2://', 's3://')
         download_via_awscli = ('AWS_SHARED_CREDENTIALS_FILE='
                                f'{cloudflare.R2_CREDENTIALS_PATH} '
-                               f'aws s3 cp s3://{source} {destination} '
+                               f'aws s3 cp {source} {destination} '
                                f'--endpoint {endpoint_url} '
                                f'--profile={cloudflare.R2_PROFILE_NAME}')
 
@@ -223,8 +226,76 @@ def get_storage_from_path(url: str) -> CloudStorage:
     return _REGISTRY[result.scheme]
 
 
+class IBMCosCloudStorage(CloudStorage):
+    """IBM Cloud Storage."""
+    # install rclone if package isn't already installed
+    _GET_RCLONE = [
+        'rclone version >/dev/null 2>&1 || '
+        'curl https://rclone.org/install.sh | sudo bash',
+    ]
+
+    def is_directory(self, url: str) -> bool:
+        """Returns whether IBM COS bucket's 'url' is a directory.
+
+        In cloud object stores, a "directory" refers to a regular object whose
+        name is a prefix of other objects.
+        """
+
+        bucket_name, path, region = data_utils.split_cos_path(url)
+        s3 = ibm.get_cos_resource(region)
+        bucket = s3.Bucket(bucket_name)
+
+        num_objects = 0
+        for obj in bucket.objects.filter(Prefix=path):
+            num_objects += 1
+            if obj.key == path:
+                return False
+            # If there are more than 1 object in filter, then it is a directory
+            if num_objects == 3:
+                return True
+
+        # A directory with few or no items
+        return True
+
+    def _get_rclone_sync_command(self, source: str, destination: str):
+        bucket_name, data_path, bucket_region = data_utils.split_cos_path(
+            source)
+        bucket_rclone_profile = Rclone.generate_rclone_bucket_profile_name(
+            bucket_name, Rclone.RcloneClouds.IBM)
+        data_path_in_bucket = bucket_name + data_path
+        rclone_config_data = Rclone.get_rclone_config(bucket_name,
+                                                      Rclone.RcloneClouds.IBM,
+                                                      bucket_region)
+        # configure_rclone stores bucket profile in remote cluster's rclone.conf
+        configure_rclone = (
+            f' mkdir -p ~/.config/rclone/ &&'
+            f' echo "{rclone_config_data}">> {Rclone.RCLONE_CONFIG_PATH}')
+        download_via_rclone = (
+            'rclone copy '
+            f'{bucket_rclone_profile}:{data_path_in_bucket} {destination}')
+
+        all_commands = list(self._GET_RCLONE)
+        all_commands.append(configure_rclone)
+        all_commands.append(download_via_rclone)
+        return ' && '.join(all_commands)
+
+    def make_sync_dir_command(self, source: str, destination: str) -> str:
+        """Downloads a directory from 'source' bucket to remote vm
+          at 'destination' using rclone."""
+        return self._get_rclone_sync_command(source, destination)
+
+    def make_sync_file_command(self, source: str, destination: str) -> str:
+        """Downloads a file from 'source' bucket to remote vm
+          at 'destination' using rclone."""
+
+        # underlying rclone command is the same for dirs and files.
+        return self.make_sync_dir_command(source, destination)
+
+
+# Maps bucket's URIs prefix(scheme) to its corresponding storage class
 _REGISTRY = {
     'gs': GcsCloudStorage(),
     's3': S3CloudStorage(),
-    'r2': R2CloudStorage()
+    'r2': R2CloudStorage(),
+    'cos': IBMCosCloudStorage(),
 }
diff --git a/sky/clouds/ibm.py b/sky/clouds/ibm.py
index a4e4f2753..b666fdc83 100644
--- a/sky/clouds/ibm.py
+++ b/sky/clouds/ibm.py
@@ -1,6 +1,6 @@
 """IBM Web Services."""
+import colorama
 import os
-import yaml
 import json
 import typing
 from typing import Any, Dict, Iterator, List, Optional, Tuple
@@ -374,22 +374,32 @@ class IBM(clouds.Cloud):
     @classmethod
     def check_credentials(cls) -> Tuple[bool, Optional[str]]:
         """Checks if the user has access credentials to this cloud."""
-        # IBM-TODO - create a configuration script.
+
         required_fields = ['iam_api_key', 'resource_group_id']
+        ibm_cos_fields = ['access_key_id', 'secret_access_key']
         help_str = ('    Store your API key and Resource Group id '
                     f'in {CREDENTIAL_FILE} in the following format:\n'
                     '      iam_api_key: <IAM_API_KEY>\n'
                     '      resource_group_id: <RESOURCE_GROUP_ID>')
+        base_config = ibm.read_credential_file()
 
-        base_config = _read_credential_file()
         if not base_config:
             return (False, 'Missing credential file at '
                     f'{os.path.expanduser(CREDENTIAL_FILE)}.\n' + help_str)
-
-        for field in required_fields:
-            if field not in base_config:
-                return (False, f'Missing field "{field}" in '
-                        f'{os.path.expanduser(CREDENTIAL_FILE)}.\n' + help_str)
+        # TODO(IBM) update when issue #1943 is resolved.
+        if set(ibm_cos_fields) - set(base_config):
+            logger.debug(f'{colorama.Fore.RED}IBM Storage is missing the '
+                         'following fields in '
+                         f'{os.path.expanduser(CREDENTIAL_FILE)} to function: '
+                         f"""{", ".join(list(
+                            set(ibm_cos_fields) - set(base_config)))}"""
+                         f'{colorama.Style.RESET_ALL}')
+
+        if set(required_fields) - set(base_config):
+            return (
+                False, f'Missing field(s): '
+                f'{", ".join(list(set(required_fields) - set(base_config)))} '
+                f'in {os.path.expanduser(CREDENTIAL_FILE)}.\n{help_str}')
 
         # verifies ability of user to create a client,
         # e.g. bad API KEY.
@@ -468,19 +478,10 @@ class IBM(clouds.Cloud):
         return [status_map[instance['status']] for instance in instances]
 
 
-def _read_credential_file():
-    try:
-        with open(os.path.expanduser(CREDENTIAL_FILE), 'r',
-                  encoding='utf-8') as f:
-            return yaml.safe_load(f)
-    except FileNotFoundError:
-        return False
-
-
 def get_cred_file_field(field, default_val=None) -> str:
     """returns a the value of a field from the user's
      credentials file if exists, else default_val"""
-    base_config = _read_credential_file()
+    base_config = ibm.read_credential_file()
     if not base_config:
         raise FileNotFoundError('Missing '
                                 f'credential file at {CREDENTIAL_FILE}')
diff --git a/sky/data/data_utils.py b/sky/data/data_utils.py
index eeb827df4..0882f755e 100644
--- a/sky/data/data_utils.py
+++ b/sky/data/data_utils.py
@@ -1,14 +1,20 @@
 """Miscellaneous Utils for Sky Data
 """
+import concurrent.futures
+from enum import Enum
 from multiprocessing import pool
 import os
 import subprocess
+import textwrap
 from typing import Any, Callable, Dict, List, Optional, Tuple
 import urllib.parse
+import re
+
+from filelock import FileLock
 
 from sky import exceptions
 from sky import sky_logging
-from sky.adaptors import aws, gcp, cloudflare
+from sky.adaptors import aws, gcp, cloudflare, ibm
 from sky.utils import ux_utils
 
 Client = Any
@@ -52,6 +58,35 @@ def split_r2_path(r2_path: str) -> Tuple[str, str]:
     return bucket, key
 
 
+def split_cos_path(s3_path: str) -> Tuple[str, str, str]:
+    """returns extracted region, bucket name and bucket path to data
+        from the specified cos bucket's url.
+        url expected format: "cos://region/bucket_name/optional_data_path"
+
+        Raises:
+          ValueError if s3_path isn't a valid IBM COS bucket URI."""
+
+    # regex pattern: 3rd group differs from 2nd by accepting '/'
+    pattern_region_bucket_data = r'cos://([-\w]+)/([-\w]+)(.*)'
+    try:
+        match = re.match(pattern_region_bucket_data, s3_path)
+    except TypeError:
+        # URI provided isn't a string
+        raise ValueError(f'URI received: {s3_path} is not valid.')  # pylint: disable=raise-missing-from
+    if match:
+        region, bucket_name, data_path = match.group(1), match.group(
+            2), match.group(3)
+    else:
+        raise ValueError(
+            f'Bucket URI received: {s3_path} does not match expected pattern '
+            'of "cos://region/bucket_name/optional_data_path"')
+
+    if region not in get_cos_regions():
+        raise ValueError('region missing/invalid in IBM COS URI.')
+
+    return bucket_name, data_path, region
+
+
 def create_s3_client(region: str = 'us-east-2') -> Client:
     """Helper method that connects to Boto3 client for S3 Bucket
 
@@ -105,6 +140,82 @@ def verify_r2_bucket(name: str) -> bool:
     return bucket in r2.buckets.all()
 
 
+def verify_ibm_cos_bucket(name: str) -> bool:
+    """Helper method that checks if the cos bucket exists
+
+    Args:
+      name: str; Name of a COS Bucket (without cos://region/ prefix)
+    """
+    return get_ibm_cos_bucket_region(name) != ''
+
+
+def _get_ibm_cos_bucket_region(region, bucket_name):
+    """helper function of get_ibm_cos_bucket_region
+
+    returns region of bucket if exists in a region,
+    else returns an empty string.
+    located outside of get_ibm_cos_bucket_region since
+    inner function aren't pickable for processes.
+
+    Args:
+        region (str): region to search the bucket in.
+        bucket_name (str): name of bucket in search.
+
+    Returns:
+        str: region or ''
+    """
+    try:
+        # reinitialize a client to search in different regions
+        tmp_client = ibm.get_cos_client(region)
+        tmp_client.head_bucket(Bucket=bucket_name)
+        return region
+    except ibm.ibm_botocore.exceptions.ClientError as e:  # type: ignore[union-attr] # pylint: disable=line-too-long
+        if e.response['Error']['Code'] == '404':
+            logger.debug(f'bucket {bucket_name} was not found '
+                         f'in {region}')
+        # Failed to connect to bucket, e.g. owned by different user
+        elif e.response['Error']['Code'] == '403':
+            raise exceptions.StorageBucketGetError()
+        else:
+            raise e
+    return ''
+
+
+def get_ibm_cos_bucket_region(bucket_name: str) -> str:
+    """
+    Returns the region of the bucket if exists,
+        otherwise returns empty string.
+
+    Args:
+        bucket_name (str): name of IBM COS bucket.
+
+    Raises:
+        exceptions.StorageBucketGetError if failed to connect to bucket.
+
+    Returns:
+        str: region of bucket if bucket exists, else empty string.
+    """
+
+    # parallel lookup on bucket storage region
+    # Using threads would be more efficient, but raises a leaked semaphore
+    # warning on Mac machines when guarding cos client with a thread lock
+    # nested within the process lock. Source:
+    # https://github.com/skypilot-org/skypilot/pull/1966#issuecomment-1646992938
+    with concurrent.futures.ProcessPoolExecutor() as executor:
+        futures = []
+        for region_scanned in get_cos_regions():
+            future = executor.submit(_get_ibm_cos_bucket_region, region_scanned,
+                                     bucket_name)
+            futures.append(future)
+
+        for future in concurrent.futures.as_completed(futures):
+            region = future.result()
+            if region:
+                logger.debug(f'bucket {bucket_name} was found in {region}')
+                return region
+    return ''
+
+
 def is_cloud_store_url(url):
     result = urllib.parse.urlsplit(url)
     # '' means non-cloud URLs.
@@ -263,3 +374,192 @@ def run_upload_cli(command: str, access_denied_message: str, bucket_name: str):
                 raise exceptions.StorageUploadError(
                     f'Upload to bucket failed for store {bucket_name}. '
                     'Please check the logs.')
+
+
+def get_cos_regions() -> List[str]:
+    return [
+        'us-south', 'us-east', 'eu-de', 'eu-gb', 'eu-es', 'ca-tor', 'au-syd',
+        'br-sao', 'jp-osa', 'jp-tok'
+    ]
+
+
+class Rclone():
+    """
+    Static class implementing common utilities of rclone without
+    rclone sdk.
+
+    Storage providers supported by rclone are required to:
+    - list their rclone profile prefix in RcloneClouds
+    - implement configuration in get_rclone_config()
+    """
+
+    RCLONE_CONFIG_PATH = '~/.config/rclone/rclone.conf'
+    _RCLONE_ABS_CONFIG_PATH = os.path.expanduser(RCLONE_CONFIG_PATH)
+
+    # Mapping of storage providers using rclone
+    # to their respective profile prefix
+    class RcloneClouds(Enum):
+        IBM = 'sky-ibm-'
+
+    @staticmethod
+    def generate_rclone_bucket_profile_name(bucket_name: str,
+                                            cloud: RcloneClouds) -> str:
+        """Returns rclone profile name for specified bucket
+
+        Args:
+            bucket_name (str): name of bucket
+            cloud (RcloneClouds): enum object of storage provider
+                supported via rclone
+        """
+        try:
+            return cloud.value + bucket_name
+        except AttributeError as e:
+            with ux_utils.print_exception_no_traceback():
+                raise ValueError(f'Value: {cloud} isn\'t a member of '
+                                 'Rclone.RcloneClouds') from e
+
+    @staticmethod
+    def get_rclone_config(bucket_name: str, cloud: RcloneClouds,
+                          region: str) -> str:
+        bucket_rclone_profile = Rclone.generate_rclone_bucket_profile_name(
+            bucket_name, cloud)
+        if cloud is Rclone.RcloneClouds.IBM:
+            access_key_id, secret_access_key = ibm.get_hmac_keys()
+            config_data = textwrap.dedent(f"""\
+                [{bucket_rclone_profile}]
+                type = s3
+                provider = IBMCOS
+                access_key_id = {access_key_id}
+                secret_access_key = {secret_access_key}
+                region = {region}
+                endpoint = s3.{region}.cloud-object-storage.appdomain.cloud
+                location_constraint = {region}-smart
+                acl = private
+                """)
+        else:
+            with ux_utils.print_exception_no_traceback():
+                raise NotImplementedError('No rclone configuration builder was '
+                                          f'implemented for cloud: {cloud}.')
+        return config_data
+
+    @staticmethod
+    def store_rclone_config(bucket_name: str, cloud: RcloneClouds,
+                            region: str) -> str:
+        """Creates a configuration files for rclone - used for
+        bucket syncing and mounting """
+
+        rclone_config_path = Rclone._RCLONE_ABS_CONFIG_PATH
+        config_data = Rclone.get_rclone_config(bucket_name, cloud, region)
+
+        # Raise exception if rclone isn't installed
+        try:
+            subprocess.run('rclone version',
+                           shell=True,
+                           check=True,
+                           stdout=subprocess.PIPE)
+        except subprocess.CalledProcessError:
+            with ux_utils.print_exception_no_traceback():
+                raise exceptions.StorageError(
+                    'rclone wasn\'t detected. '
+                    'Consider installing via: '
+                    '"curl https://rclone.org/install.sh '
+                    '| sudo bash" ') from None
+
+        # create ~/.config/rclone/ if doesn't exist
+        os.makedirs(os.path.dirname(rclone_config_path), exist_ok=True)
+        # create rclone.conf if doesn't exist
+        if not os.path.isfile(rclone_config_path):
+            open(rclone_config_path, 'w').close()
+
+        # write back file without profile: [bucket_name]
+        # to which the new bucket profile is appended
+        with FileLock(rclone_config_path + '.lock'):
+            profiles_to_keep = Rclone._remove_bucket_profile_rclone(
+                bucket_name, cloud)
+            with open(f'{rclone_config_path}', 'w') as file:
+                if profiles_to_keep:
+                    file.writelines(profiles_to_keep)
+                    if profiles_to_keep[-1].strip():
+                        # add a new line to config_data
+                        # if last file line contains data
+                        config_data += '\n'
+                file.write(config_data)
+
+        return config_data
+
+    @staticmethod
+    def get_region_from_rclone(bucket_name: str, cloud: RcloneClouds) -> str:
+        """Returns region field of the specified bucket in rclone.conf
+         if bucket exists, else empty string"""
+        bucket_rclone_profile = Rclone.generate_rclone_bucket_profile_name(
+            bucket_name, cloud)
+        with open(Rclone._RCLONE_ABS_CONFIG_PATH) as file:
+            bucket_profile_found = False
+            for line in file:
+                if line.lstrip().startswith('#'):  # skip user's comments.
+                    continue
+                if line.strip() == f'[{bucket_rclone_profile}]':
+                    bucket_profile_found = True
+                elif bucket_profile_found and line.startswith('region'):
+                    return line.split('=')[1].strip()
+                elif bucket_profile_found and line.startswith('['):
+                    # for efficiency stop if we've searched past the
+                    # requested bucket profile with no match
+                    return ''
+        # segment for bucket and/or region field for bucket wasn't found
+        return ''
+
+    @staticmethod
+    def delete_rclone_bucket_profile(bucket_name: str, cloud: RcloneClouds):
+        """Deletes specified bucket profile for rclone.conf"""
+        bucket_rclone_profile = Rclone.generate_rclone_bucket_profile_name(
+            bucket_name, cloud)
+        rclone_config_path = Rclone._RCLONE_ABS_CONFIG_PATH
+
+        if not os.path.isfile(rclone_config_path):
+            logger.warning(
+                'Failed to locate "rclone.conf" while '
+                f'trying to delete rclone profile: {bucket_rclone_profile}')
+            return
+
+        with FileLock(rclone_config_path + '.lock'):
+            profiles_to_keep = Rclone._remove_bucket_profile_rclone(
+                bucket_name, cloud)
+
+            # write back file without profile: [bucket_rclone_profile]
+            with open(f'{rclone_config_path}', 'w') as file:
+                file.writelines(profiles_to_keep)
+
+    @staticmethod
+    def _remove_bucket_profile_rclone(bucket_name: str,
+                                      cloud: RcloneClouds) -> List[str]:
+        """Returns rclone profiles without profiles matching
+          [profile_prefix+bucket_name]"""
+        bucket_rclone_profile = Rclone.generate_rclone_bucket_profile_name(
+            bucket_name, cloud)
+        rclone_config_path = Rclone._RCLONE_ABS_CONFIG_PATH
+
+        with open(f'{rclone_config_path}', 'r') as file:
+            lines = file.readlines()  # returns a list of the file's lines
+            # delete existing bucket profile matching:
+            # '[profile_prefix+bucket_name]'
+            lines_to_keep = []  # lines to write back to file
+            # while skip_lines is True avoid adding lines to lines_to_keep
+            skip_lines = False
+
+        for line in lines:
+            if line.lstrip().startswith('#') and not skip_lines:
+                # keep user comments only if they aren't under
+                # a profile we are discarding
+                lines_to_keep.append(line)
+            elif f'[{bucket_rclone_profile}]' in line:
+                skip_lines = True
+            elif skip_lines:
+                if '[' in line:
+                    # former profile segment ended, new one begins
+                    skip_lines = False
+                    lines_to_keep.append(line)
+            else:
+                lines_to_keep.append(line)
+
+        return lines_to_keep
diff --git a/sky/data/storage.py b/sky/data/storage.py
index 66276e0f2..9634ce774 100644
--- a/sky/data/storage.py
+++ b/sky/data/storage.py
@@ -15,11 +15,13 @@ from sky import clouds
 from sky.adaptors import aws
 from sky.adaptors import gcp
 from sky.adaptors import cloudflare
+from sky.adaptors import ibm
 from sky.backends import backend_utils
 from sky.utils import schemas
 from sky.data import data_transfer
 from sky.data import data_utils
 from sky.data import mounting_utils
+from sky.data.data_utils import Rclone
 from sky.data import storage_utils
 from sky import exceptions
 from sky import global_user_state
@@ -44,7 +46,9 @@ SourceType = Union[Path, List[Path]]
 # TODO(Doyoung): need to add clouds.CLOUDFLARE() to support
 # R2 to be an option as preferred store type
 STORE_ENABLED_CLOUDS: List[str] = [
-    str(clouds.AWS()), str(clouds.GCP()), cloudflare.NAME
+    str(clouds.AWS()),
+    str(clouds.GCP()),
+    str(clouds.IBM()), cloudflare.NAME
 ]
 
 # Maximum number of concurrent rsync upload processes
@@ -82,6 +86,7 @@ class StoreType(enum.Enum):
     GCS = 'GCS'
     AZURE = 'AZURE'
     R2 = 'R2'
+    IBM = 'IBM'
 
     @classmethod
     def from_cloud(cls, cloud: clouds.Cloud) -> 'StoreType':
@@ -91,6 +96,8 @@ class StoreType(enum.Enum):
             return StoreType.GCS
         elif isinstance(cloud, clouds.Azure):
             return StoreType.AZURE
+        elif isinstance(cloud, clouds.IBM):
+            return StoreType.IBM
 
         raise ValueError(f'Unsupported cloud for StoreType: {cloud}')
 
@@ -102,6 +109,8 @@ class StoreType(enum.Enum):
             return StoreType.GCS
         elif isinstance(store, R2Store):
             return StoreType.R2
+        elif isinstance(store, IBMCosStore):
+            return StoreType.IBM
         else:
             with ux_utils.print_exception_no_traceback():
                 raise ValueError(f'Unknown store type: {store}')
@@ -117,6 +126,8 @@ def get_storetype_from_cloud(cloud: clouds.Cloud) -> StoreType:
         return StoreType.S3
     elif isinstance(cloud, clouds.GCP):
         return StoreType.GCS
+    elif isinstance(cloud, clouds.IBM):
+        return StoreType.IBM
     elif isinstance(cloud, clouds.Azure):
         with ux_utils.print_exception_no_traceback():
             raise ValueError('Azure Blob Storage is not supported yet.')
@@ -139,6 +150,8 @@ def get_store_prefix(storetype: StoreType) -> str:
     # R2 storages use 's3://' as a prefix for various aws cli commands
     elif storetype == StoreType.R2:
         return 's3://'
+    elif storetype == StoreType.IBM:
+        return 'cos://'
     elif storetype == StoreType.AZURE:
         with ux_utils.print_exception_no_traceback():
             raise ValueError('Azure Blob Storage is not supported yet.')
@@ -463,6 +476,11 @@ class Storage(object):
                         s_metadata,
                         source=self.source,
                         sync_on_reconstruction=self.sync_on_reconstruction)
+                elif s_type == StoreType.IBM:
+                    store = IBMCosStore.from_metadata(
+                        s_metadata,
+                        source=self.source,
+                        sync_on_reconstruction=self.sync_on_reconstruction)
                 else:
                     with ux_utils.print_exception_no_traceback():
                         raise ValueError(f'Unknown store type: {s_type}')
@@ -501,6 +519,8 @@ class Storage(object):
                         self.add_store(StoreType.GCS)
                     elif self.source.startswith('r2://'):
                         self.add_store(StoreType.R2)
+                    elif self.source.startswith('cos://'):
+                        self.add_store(StoreType.IBM)
 
     @staticmethod
     def _validate_source(
@@ -581,12 +601,18 @@ class Storage(object):
                             'using a bucket by writing <destination_path>: '
                             f'{source} in the file_mounts section of your YAML')
                 is_local_source = True
-            elif split_path.scheme in ['s3', 'gs', 'r2']:
+            elif split_path.scheme in ['s3', 'gs', 'r2', 'cos']:
                 is_local_source = False
                 # Storage mounting does not support mounting specific files from
                 # cloud store - ensure path points to only a directory
                 if mode == StorageMode.MOUNT:
-                    if split_path.path.strip('/') != '':
+                    if ((not split_path.scheme == 'cos' and
+                         split_path.path.strip('/') != '') or
+                        (split_path.scheme == 'cos' and
+                         not re.match(r'^/[-\w]+(/\s*)?$', split_path.path))):
+                        # regex allows split_path.path to include /bucket
+                        # or /bucket/optional_whitespaces while considering
+                        # cos URI's regions (cos://region/bucket_name)
                         with ux_utils.print_exception_no_traceback():
                             raise exceptions.StorageModeError(
                                 'MOUNT mode does not support'
@@ -598,7 +624,7 @@ class Storage(object):
                 with ux_utils.print_exception_no_traceback():
                     raise exceptions.StorageSourceError(
                         f'Supported paths: local, s3://, gs://, '
-                        f'r2://. Got: {source}')
+                        f'r2://, cos://. Got: {source}')
         return source, is_local_source
 
     def _validate_storage_spec(self, name: Optional[str]) -> None:
@@ -615,7 +641,7 @@ class Storage(object):
             """
             prefix = name.split('://')[0]
             prefix = prefix.lower()
-            if prefix in ['s3', 'gs', 'r2']:
+            if prefix in ['s3', 'gs', 'r2', 'cos']:
                 with ux_utils.print_exception_no_traceback():
                     raise exceptions.StorageNameError(
                         'Prefix detected: `name` cannot start with '
@@ -663,7 +689,11 @@ class Storage(object):
                 else:
                     assert isinstance(source, str)
                     # Set name to source bucket name and continue
-                    name = urllib.parse.urlsplit(source).netloc
+                    if source.startswith('cos://'):
+                        # cos url requires custom parsing
+                        name = data_utils.split_cos_path(source)[0]
+                    else:
+                        name = urllib.parse.urlsplit(source).netloc
                     assert name is not None, source
                     self.name = name
                     return
@@ -692,7 +722,7 @@ class Storage(object):
         add it to Storage.
 
         Args:
-          store_type: StoreType; Type of the storage [S3, GCS, AZURE, R2]
+          store_type: StoreType; Type of the storage [S3, GCS, AZURE, R2, IBM]
         """
         if isinstance(store_type, str):
             store_type = StoreType(store_type)
@@ -708,6 +738,8 @@ class Storage(object):
             store_cls = GcsStore
         elif store_type == StoreType.R2:
             store_cls = R2Store
+        elif store_type == StoreType.IBM:
+            store_cls = IBMCosStore
         else:
             with ux_utils.print_exception_no_traceback():
                 raise exceptions.StorageSpecError(
@@ -930,6 +962,13 @@ class S3Store(AbstractStore):
                 assert data_utils.verify_r2_bucket(self.name), (
                     f'Source specified as {self.source}, a R2 bucket. ',
                     'R2 Bucket should exist.')
+            elif self.source.startswith('cos://'):
+                assert self.name == data_utils.split_cos_path(self.source)[0], (
+                    'COS Bucket is specified as path, the name should be '
+                    'the same as COS bucket.')
+                assert data_utils.verify_ibm_cos_bucket(self.name), (
+                    f'Source specified as {self.source}, a COS bucket. ',
+                    'COS Bucket should exist.')
         # Validate name
         self.name = self.validate_name(self.name)
 
@@ -1320,6 +1359,15 @@ class GcsStore(AbstractStore):
                     assert data_utils.verify_r2_bucket(self.name), (
                         f'Source specified as {self.source}, a R2 bucket. ',
                         'R2 Bucket should exist.')
+                elif self.source.startswith('cos://'):
+                    assert self.name == data_utils.split_cos_path(
+                        self.source
+                    )[0], (
+                        'COS Bucket is specified as path, the name should be '
+                        'the same as COS bucket.')
+                    assert data_utils.verify_ibm_cos_bucket(self.name), (
+                        f'Source specified as {self.source}, a COS bucket. ',
+                        'COS Bucket should exist.')
         # Validate name
         self.name = self.validate_name(self.name)
         # Check if the storage is enabled
@@ -1744,6 +1792,13 @@ class R2Store(AbstractStore):
                 assert self.name == data_utils.split_r2_path(self.source)[0], (
                     'R2 Bucket is specified as path, the name should be '
                     'the same as R2 bucket.')
+            elif self.source.startswith('cos://'):
+                assert self.name == data_utils.split_cos_path(self.source)[0], (
+                    'IBM COS Bucket is specified as path, the name should be '
+                    'the same as COS bucket.')
+                assert data_utils.verify_ibm_cos_bucket(self.name), (
+                    f'Source specified as {self.source}, a COS bucket. ',
+                    'COS Bucket should exist.')
         # Validate name
         self.name = S3Store.validate_name(self.name)
         # Check if the storage is enabled
@@ -2057,3 +2112,389 @@ class R2Store(AbstractStore):
         while data_utils.verify_r2_bucket(bucket_name):
             time.sleep(0.1)
         return True
+
+
+class IBMCosStore(AbstractStore):
+    """IBMCosStore inherits from Storage Object and represents the backend
+    for COS buckets.
+    """
+    _ACCESS_DENIED_MESSAGE = 'Access Denied'
+
+    def __init__(self,
+                 name: str,
+                 source: str,
+                 region: Optional[str] = 'us-east',
+                 is_sky_managed: Optional[bool] = None,
+                 sync_on_reconstruction: bool = True):
+        self.client: 'storage.Client'
+        self.bucket: 'StorageHandle'
+        super().__init__(name, source, region, is_sky_managed,
+                         sync_on_reconstruction)
+        self.bucket_rclone_profile = \
+          Rclone.generate_rclone_bucket_profile_name(
+            self.name, Rclone.RcloneClouds.IBM)
+
+    def _validate(self):
+        if self.source is not None and isinstance(self.source, str):
+            if self.source.startswith('s3://'):
+                assert self.name == data_utils.split_s3_path(self.source)[0], (
+                    'S3 Bucket is specified as path, the name should be the'
+                    ' same as S3 bucket.')
+                assert data_utils.verify_s3_bucket(self.name), (
+                    f'Source specified as {self.source}, a S3 bucket. ',
+                    'S3 Bucket should exist.')
+            elif self.source.startswith('gs://'):
+                assert self.name == data_utils.split_gcs_path(self.source)[0], (
+                    'GCS Bucket is specified as path, the name should be '
+                    'the same as GCS bucket.')
+                assert data_utils.verify_gcs_bucket(self.name), (
+                    f'Source specified as {self.source}, a GCS bucket. ',
+                    'GCS Bucket should exist.')
+            elif self.source.startswith('r2://'):
+                assert self.name == data_utils.split_r2_path(self.source)[0], (
+                    'R2 Bucket is specified as path, the name should be '
+                    'the same as R2 bucket.')
+                assert data_utils.verify_r2_bucket(self.name), (
+                    f'Source specified as {self.source}, a R2 bucket. ',
+                    'R2 Bucket should exist.')
+            elif self.source.startswith('cos://'):
+                assert self.name == data_utils.split_cos_path(self.source)[0], (
+                    'COS Bucket is specified as path, the name should be '
+                    'the same as COS bucket.')
+        # Validate name
+        self.name = IBMCosStore.validate_name(self.name)
+
+    @classmethod
+    def validate_name(cls, name) -> str:
+        """Validates the name of a COS bucket.
+
+        Rules source: https://ibm.github.io/ibm-cos-sdk-java/com/ibm/cloud/objectstorage/services/s3/model/Bucket.html  # pylint: disable=line-too-long
+        """
+
+        def _raise_no_traceback_name_error(err_str):
+            with ux_utils.print_exception_no_traceback():
+                raise exceptions.StorageNameError(err_str)
+
+        if name is not None and isinstance(name, str):
+            if not 3 <= len(name) <= 63:
+                _raise_no_traceback_name_error(
+                    f'Invalid store name: {name} must be between 3 (min) '
+                    'and 63 (max) characters long.')
+
+            # Check for valid characters and start/end with a letter or number
+            pattern = r'^[a-z0-9][-a-z0-9.]*[a-z0-9]$'
+            if not re.match(pattern, name):
+                _raise_no_traceback_name_error(
+                    f'Invalid store name: {name} can consist only of '
+                    'lowercase letters, numbers, dots (.), and dashes (-). '
+                    'It must begin and end with a letter or number.')
+
+            # Check for two adjacent periods or dashes
+            if any(substring in name for substring in ['..', '--']):
+                _raise_no_traceback_name_error(
+                    f'Invalid store name: {name} must not contain '
+                    'two adjacent periods/dashes')
+
+            # Check for IP address format
+            ip_pattern = r'^(?:\d{1,3}\.){3}\d{1,3}$'
+            if re.match(ip_pattern, name):
+                _raise_no_traceback_name_error(
+                    f'Invalid store name: {name} must not be formatted as '
+                    'an IP address (for example, 192.168.5.4).')
+
+            if any(substring in name for substring in ['.-', '-.']):
+                _raise_no_traceback_name_error(
+                    f'Invalid store name: {name} must '
+                    'not allow substrings: ".-", "-." .')
+        else:
+            _raise_no_traceback_name_error('Store name must be specified.')
+        return name
+
+    def initialize(self):
+        """Initializes the cos store object on the cloud.
+
+        Initialization involves fetching bucket if exists, or creating it if
+        it does not.
+
+        Raises:
+          StorageBucketCreateError: If bucket creation fails
+          StorageBucketGetError: If fetching existing bucket fails
+          StorageInitError: If general initialization fails.
+        """
+        self.client = ibm.get_cos_client(self.region)
+        self.s3_resource = ibm.get_cos_resource(self.region)
+        self.bucket, is_new_bucket = self._get_bucket()
+        if self.is_sky_managed is None:
+            # If is_sky_managed is not specified, then this is a new storage
+            # object (i.e., did not exist in global_user_state) and we should
+            # set the is_sky_managed property.
+            # If is_sky_managed is specified, then we take no action.
+            self.is_sky_managed = is_new_bucket
+
+    def upload(self):
+        """Uploads files from local machine to bucket.
+
+        Upload must be called by the Storage handler - it is not called on
+        Store initialization.
+
+        Raises:
+            StorageUploadError: if upload fails.
+        """
+        try:
+            if isinstance(self.source, list):
+                self.batch_ibm_rsync(self.source, create_dirs=True)
+            elif self.source is not None:
+                if self.source.startswith('cos://'):
+                    # cos bucket used as a dest, can't be used as source.
+                    pass
+                elif self.source.startswith('s3://'):
+                    raise Exception('IBM COS currently not supporting'
+                                    'data transfers between COS and S3')
+                elif self.source.startswith('gs://'):
+                    raise Exception('IBM COS currently not supporting'
+                                    'data transfers between COS and GS')
+                elif self.source.startswith('r2://'):
+                    raise Exception('IBM COS currently not supporting'
+                                    'data transfers between COS and r2')
+                else:
+                    self.batch_ibm_rsync([self.source])
+
+        except Exception as e:
+            raise exceptions.StorageUploadError(
+                f'Upload failed for store {self.name}') from e
+
+    def delete(self) -> None:
+        self._delete_cos_bucket()
+        logger.info(f'{colorama.Fore.GREEN}Deleted COS bucket {self.name}.'
+                    f'{colorama.Style.RESET_ALL}')
+
+    def get_handle(self) -> StorageHandle:
+        return self.s3_resource.Bucket(self.name)
+
+    def batch_ibm_rsync(self,
+                        source_path_list: List[Path],
+                        create_dirs: bool = False) -> None:
+        """Invokes rclone copy to batch upload a list of local paths to cos
+
+        Since rclone does not support batch operations, we construct
+        multiple commands to be run in parallel.
+
+        Args:
+            source_path_list: List of paths to local files or directories
+            create_dirs: If the local_path is a directory and this is set to
+                False, the contents of the directory are directly uploaded to
+                root of the bucket. If the local_path is a directory and this is
+                set to True, the directory is created in the bucket root and
+                contents are uploaded to it.
+        """
+
+        def get_dir_sync_command(src_dir_path, dest_dir_name) -> str:
+            """returns an rclone command that copies a complete folder
+              from 'src_dir_path' to bucket/'dest_dir_name'.
+
+            `rclone copy` copies files from source path to target.
+            files with identical names at won't be copied over, unless
+            their modification date is more recent.
+            works similarly to `aws sync` (without --delete).
+
+            Args:
+                src_dir_path (str): local source path from which to copy files.
+                dest_dir_name (str): remote target path files are copied to.
+
+            Returns:
+                str: bash command using rclone to sync files. Executed remotely.
+            """
+
+            # .git directory is excluded from the sync
+            # wrapping src_dir_path with "" to support path with spaces
+            sync_command = (
+                'rclone copy --exclude ".git/*" '
+                f'"{src_dir_path}" '
+                f'{self.bucket_rclone_profile}:{self.name}/{dest_dir_name}')
+            return sync_command
+
+        def get_file_sync_command(base_dir_path, file_names) -> str:
+            """returns an rclone command that copies files: 'file_names'
+               from base directory: `base_dir_path` to bucket.
+
+            `rclone copy` copies files from source path to target.
+            files with identical names at won't be copied over, unless
+            their modification date is more recent.
+            works similarly to `aws sync` (without --delete).
+
+            Args:
+                base_dir_path (str): local path from which to copy files.
+                file_names (List): specific file names to copy.
+
+            Returns:
+                str: bash command using rclone to sync files
+            """
+
+            # wrapping file_name with "" to support spaces
+            includes = ' '.join(
+                [f'--include "{file_name}"' for file_name in file_names])
+            sync_command = ('rclone copy '
+                            f'{includes} "{base_dir_path}" '
+                            f'{self.bucket_rclone_profile}:{self.name}')
+            return sync_command
+
+        # Generate message for upload
+        if len(source_path_list) > 1:
+            source_message = f'{len(source_path_list)} paths'
+        else:
+            source_message = source_path_list[0]
+
+        with log_utils.safe_rich_status(
+                f'[bold cyan]Syncing '
+                f'[green]{source_message}[/] to '
+                f'[green]cos://{self.region}/{self.name}/[/]'):
+            data_utils.parallel_upload(
+                source_path_list,
+                get_file_sync_command,
+                get_dir_sync_command,
+                self.name,
+                self._ACCESS_DENIED_MESSAGE,
+                create_dirs=create_dirs,
+                max_concurrent_uploads=_MAX_CONCURRENT_UPLOADS)
+
+    def _get_bucket(self) -> Tuple[StorageHandle, bool]:
+        """
+        returns IBM COS bucket object if exists, otherwise creates it.
+
+        Returns:
+          StorageHandle(str): bucket name
+          bool: indicates whether a new bucket was created.
+        """
+
+        bucket_profile_name = Rclone.RcloneClouds.IBM.value + self.name
+        try:
+            bucket_region = data_utils.get_ibm_cos_bucket_region(self.name)
+        except exceptions.StorageBucketGetError as e:
+            with ux_utils.print_exception_no_traceback():
+                command = f'rclone lsd {bucket_profile_name}: '
+                raise exceptions.StorageBucketGetError(
+                    _BUCKET_FAIL_TO_CONNECT_MESSAGE.format(name=self.name) +
+                    f' To debug, consider running `{command}`.') from e
+
+        try:
+            uri_region = data_utils.split_cos_path(
+                self.source)[2]  # type: ignore
+        except ValueError:
+            # source isn't a cos uri
+            uri_region = ''
+
+        # bucket's region doesn't match specified region in URI
+        if bucket_region and uri_region and uri_region != bucket_region\
+              and self.sync_on_reconstruction:
+            with ux_utils.print_exception_no_traceback():
+                raise exceptions.StorageBucketGetError(
+                    f'Bucket {self.name} exists in '
+                    f'region {bucket_region}, '
+                    f'but URI specified region {uri_region}.')
+
+        if not bucket_region and uri_region:
+            # bucket doesn't exist but source is a bucket URI
+            with ux_utils.print_exception_no_traceback():
+                raise exceptions.StorageBucketGetError(
+                    'Attempted to connect to a non-existent bucket: '
+                    f'{self.name} by providing URI. Consider using '
+                    '`rclone lsd <remote>` on relevant remotes returned '
+                    'via `rclone listremotes` to debug.')
+
+        Rclone.store_rclone_config(
+            self.name,
+            Rclone.RcloneClouds.IBM,
+            self.region,  # type: ignore
+        )
+        if not bucket_region and self.sync_on_reconstruction:
+            # bucket doesn't exist
+            return self._create_cos_bucket(self.name, self.region), True
+        else:
+            # bucket exists
+            return self.s3_resource.Bucket(self.name), False
+
+    def _download_file(self, remote_path: str, local_path: str) -> None:
+        """Downloads file from remote to local on s3 bucket
+        using the boto3 API
+
+        Args:
+          remote_path: str; Remote path on S3 bucket
+          local_path: str; Local path on user's device
+        """
+        self.client.download_file(self.name, local_path, remote_path)
+
+    def mount_command(self, mount_path: str) -> str:
+        """Returns the command to mount the bucket to the mount_path.
+
+        Uses rclone to mount the bucket.
+        Source: https://github.com/rclone/rclone
+
+        Args:
+          mount_path: str; Path to mount the bucket to.
+        """
+        rclone_config_data = Rclone.get_rclone_config(
+            self.bucket.name,
+            Rclone.RcloneClouds.IBM,
+            self.region,  # type: ignore
+        )
+        # pylint: disable=line-too-long
+        # creates a fusermount soft link on older (<22) Ubuntu systems for rclone's mount utility.
+        create_fuser3_soft_link = '[ ! -f /bin/fusermount3 ] && sudo ln -s /bin/fusermount /bin/fusermount3 || true'
+        # stores bucket profile in rclone config file at the cluster's nodes.
+        configure_rclone_profile = (
+            f'{create_fuser3_soft_link}; mkdir -p ~/.config/rclone/ && echo "{rclone_config_data}">> {Rclone.RCLONE_CONFIG_PATH}'
+        )
+        # install rclone if not installed.
+        install_cmd = 'rclone version >/dev/null 2>&1 || (curl https://rclone.org/install.sh | sudo bash)'
+        # --daemon will keep the mounting process running in the background.
+        mount_cmd = f'{configure_rclone_profile} && rclone mount {self.bucket_rclone_profile}:{self.bucket.name} {mount_path} --daemon'
+        return mounting_utils.get_mounting_command(mount_path, install_cmd,
+                                                   mount_cmd)
+
+    def _create_cos_bucket(self,
+                           bucket_name: str,
+                           region='us-east') -> StorageHandle:
+        """Creates IBM COS bucket with specific name in specific region
+
+        Args:
+          bucket_name: str; Name of bucket
+          region: str; Region name, e.g. us-east, us-south
+        Raises:
+          StorageBucketCreateError: If bucket creation fails.
+        """
+        try:
+            self.client.create_bucket(
+                Bucket=bucket_name,
+                CreateBucketConfiguration={
+                    'LocationConstraint': f'{region}-smart'
+                })
+            logger.info(f'Created IBM COS bucket {bucket_name} in {region} '
+                        f'with storage class smart tier')
+            self.bucket = self.s3_resource.Bucket(bucket_name)
+
+        except ibm.ibm_botocore.exceptions.ClientError as e:  # type: ignore[union-attr]  # pylint: disable=line-too-long
+            with ux_utils.print_exception_no_traceback():
+                raise exceptions.StorageBucketCreateError(
+                    f'Failed to create bucket: '
+                    f'{bucket_name}') from e
+
+        s3_bucket_exists_waiter = self.client.get_waiter('bucket_exists')
+        s3_bucket_exists_waiter.wait(Bucket=bucket_name)
+
+        return self.bucket
+
+    def _delete_cos_bucket(self):
+        bucket = self.s3_resource.Bucket(self.name)
+        try:
+            bucket_versioning = self.s3_resource.BucketVersioning(self.name)
+            if bucket_versioning.status == 'Enabled':
+                res = list(bucket.object_versions.delete())
+            else:
+                res = list(bucket.objects.delete())
+            logger.debug(f'Deleted bucket\'s content:\n{res}')
+            bucket.delete()
+            bucket.wait_until_not_exists()
+        except ibm.ibm_botocore.exceptions.ClientError as e:
+            if e.__class__.__name__ == 'NoSuchBucket':
+                logger.debug('bucket already removed')
+        Rclone.delete_rclone_bucket_profile(self.name, Rclone.RcloneClouds.IBM)
diff --git a/sky/setup_files/setup.py b/sky/setup_files/setup.py
index 562910d94..c9ed89f0b 100644
--- a/sky/setup_files/setup.py
+++ b/sky/setup_files/setup.py
@@ -124,11 +124,16 @@ install_requires = [
 # NOTE: Change the templates/spot-controller.yaml.j2 file if any of the
 # following packages dependencies are changed.
 aws_dependencies = [
+    # botocore does not work with urllib3>=2.0.0, accuroding to https://github.com/boto/botocore/issues/2926
+    # We have to explicitly pin the version to optimize the time for
+    # poetry install. See https://github.com/orgs/python-poetry/discussions/7937
+    'urllib3<2',
     # NOTE: this installs CLI V1. To use AWS SSO (e.g., `aws sso login`), users
     # should instead use CLI V2 which is not pip-installable. See
     # https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html.
-    'awscli',
-    'boto3',
+    'awscli>=1.27.10',
+    'botocore>=1.29.10',
+    'boto3>=1.26.1',
     # 'Crypto' module used in authentication.py for AWS.
     'pycryptodome==3.12.0',
 ]
@@ -144,7 +149,9 @@ extras_require: Dict[str, List[str]] = {
         'azure-mgmt-network'
     ],
     'gcp': ['google-api-python-client', 'google-cloud-storage'],
-    'ibm': ['ibm-cloud-sdk-core', 'ibm-vpc', 'ibm-platform-services'],
+    'ibm': [
+        'ibm-cloud-sdk-core', 'ibm-vpc', 'ibm-platform-services', 'ibm-cos-sdk'
+    ],
     'docker': ['docker'],
     'lambda': [],
     'cloudflare': aws_dependencies,
diff --git a/sky/skylet/providers/ibm/utils.py b/sky/skylet/providers/ibm/utils.py
index 7a5915ab1..b2ace9a76 100644
--- a/sky/skylet/providers/ibm/utils.py
+++ b/sky/skylet/providers/ibm/utils.py
@@ -1,7 +1,7 @@
 """holds common utility function/constants to be used by the providers."""
 
 import logging
-import os
+from pathlib import Path
 import time
 
 RAY_RECYCLABLE = "ray-recyclable"
@@ -17,8 +17,7 @@ def get_logger(caller_name):
     LOGS_FOLDER = "/tmp/connector_logs/"  # this node_provider's logs location.
     logger.setLevel(logging.DEBUG)
 
-    if not os.path.exists(LOGS_FOLDER):
-        os.mkdir(LOGS_FOLDER)
+    Path(LOGS_FOLDER).mkdir(parents=True, exist_ok=True)
     logs_path = LOGS_FOLDER + caller_name + time.strftime("%Y-%m-%d--%H-%M-%S")
     # pylint: disable=line-too-long
     file_formatter = logging.Formatter(
diff --git a/sky/spot/controller.py b/sky/spot/controller.py
index 4ecec9eb0..7bd68f5a0 100644
--- a/sky/spot/controller.py
+++ b/sky/spot/controller.py
@@ -1,6 +1,7 @@
 """Controller: handles the life cycle of a managed spot cluster (job)."""
 import argparse
 import multiprocessing
+import os
 import pathlib
 import time
 import traceback
@@ -74,8 +75,53 @@ class SpotController:
                 job_id_env_vars)
             task.update_envs(task_envs)
 
+    def _download_log_and_stream(
+            self,
+            handle: cloud_vm_ray_backend.CloudVmRayResourceHandle) -> None:
+        """Downloads and streams the logs of the latest job of a spot cluster.
+
+        We do not stream the logs from the spot cluster directly, as the
+        donwload and stream should be faster, and more robust against
+        preemptions or ssh disconnection during the streaming.
+        """
+        spot_job_logs_dir = os.path.join(constants.SKY_LOGS_DIRECTORY,
+                                         'spot_jobs')
+        os.makedirs(spot_job_logs_dir, exist_ok=True)
+        try:
+            log_dirs = self._backend.sync_down_logs(
+                handle,
+                # Download the log of the latest job.
+                # The job_id for the spot job running on the spot cluster is not
+                # necessarily 1, as it is possible that the worker node in a
+                # multi-node cluster is preempted, and we recover the spot job
+                # on the existing cluster, which leads to a larger job_id. Those
+                # job_ids all represent the same logical spot job.
+                job_ids=None,
+                local_dir=spot_job_logs_dir)
+        except exceptions.CommandError as e:
+            logger.info(f'Failed to download the logs: '
+                        f'{common_utils.format_exception(e)}')
+        else:
+            if not log_dirs:
+                logger.error('Failed to find the logs for the user program in '
+                             'the spot cluster.')
+            else:
+                log_dir = list(log_dirs.values())[0]
+                log_file = os.path.join(log_dir, 'run.log')
+
+                # Print the logs to the console.
+                try:
+                    with open(log_file) as f:
+                        print(f.read())
+                except FileNotFoundError:
+                    logger.error('Failed to find the logs for the user '
+                                 f'program at {log_file}.')
+                else:
+                    logger.info(f'\n== End of logs (ID: {self._job_id}) ==')
+
     def _run_one_task(self, task_id: int, task: 'sky.Task') -> bool:
         """Busy loop monitoring spot cluster status and handling recovery.
+
         When the task is successfully completed, this function returns True,
         and will terminate the spot cluster before returning.
 
@@ -237,13 +283,8 @@ class SpotController:
                     logger.info(
                         'The user job failed. Please check the logs below.\n'
                         f'== Logs of the user job (ID: {self._job_id}) ==\n')
-                    # TODO(zhwu): Download the logs, and stream them from the
-                    # local disk, instead of streaming them from the spot
-                    # cluster, to make it faster and more reliable.
-                    returncode = self._backend.tail_logs(
-                        handle, None, spot_job_id=self._job_id)
-                    logger.info(f'\n== End of logs (ID: {self._job_id}, '
-                                f'tail_logs returncode: {returncode}) ==')
+
+                    self._download_log_and_stream(handle)
                     spot_status_to_set = spot_state.SpotStatus.FAILED
                     if job_status == job_lib.JobStatus.FAILED_SETUP:
                         spot_status_to_set = spot_state.SpotStatus.FAILED_SETUP
diff --git a/sky/task.py b/sky/task.py
index 7e8fb37bf..39c2727c5 100644
--- a/sky/task.py
+++ b/sky/task.py
@@ -484,6 +484,8 @@ class Task:
             return clouds.AWS()
         elif self.inputs.startswith('gs:'):
             return clouds.GCP()
+        elif self.inputs.startswith('cos:'):
+            return clouds.IBM()
         else:
             with ux_utils.print_exception_no_traceback():
                 raise ValueError(f'cloud path not supported: {self.inputs}')
@@ -820,6 +822,19 @@ class Task:
                     self.update_file_mounts({
                         mnt_path: blob_path,
                     })
+                elif store_type is storage_lib.StoreType.IBM:
+                    if isinstance(storage.source,
+                                  str) and storage.source.startswith('cos://'):
+                        # source is a cos bucket's uri
+                        blob_path = storage.source
+                    else:
+                        # source is a bucket name.
+                        assert storage.name is not None, storage
+                        # extract region from rclone.conf
+                        cos_region = data_utils.Rclone.get_region_from_rclone(
+                            storage.name, data_utils.Rclone.RcloneClouds.IBM)
+                        blob_path = f'cos://{cos_region}/{storage.name}'
+                    self.update_file_mounts({mnt_path: blob_path})
                 elif store_type is storage_lib.StoreType.AZURE:
                     # TODO when Azure Blob is done: sync ~/.azure
                     raise NotImplementedError('Azure Blob not mountable yet')


Test patch:
diff --git a/.github/workflows/pytest.yml b/.github/workflows/pytest.yml
index 81a43cb49..cbb8a6214 100644
--- a/.github/workflows/pytest.yml
+++ b/.github/workflows/pytest.yml
@@ -22,7 +22,7 @@ jobs:
           - tests/test_jobs.py
           - tests/test_list_accelerators.py
           - tests/test_optimizer_dryruns.py
-          - tests/test_optimizer_random_dag.py
+          # - tests/test_optimizer_random_dag.py
           - tests/test_pycryptodome_version.py
           - tests/test_storage.py
           - tests/test_wheels.py
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index cb5932473..050e33b23 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -42,13 +42,15 @@ import pytest
 
 import sky
 from sky import global_user_state
+from sky.adaptors import ibm
+from sky.adaptors import cloudflare
+from sky.clouds import AWS, GCP, Azure
 from sky.data import data_utils
 from sky.data import storage as storage_lib
-from sky.adaptors import cloudflare
+from sky.data.data_utils import Rclone
 from sky.skylet import events
 from sky.utils import common_utils
 from sky.utils import subprocess_utils
-from sky.clouds import AWS, GCP, Azure
 
 # For uniquefying users on shared-account cloud providers. Used as part of the
 # cluster names.
@@ -868,6 +870,35 @@ def test_cloudflare_storage_mounts(generic_cloud: str):
         run_one_test(test)
 
 
+@pytest.mark.ibm
+def test_ibm_storage_mounts():
+    name = _get_cluster_name()
+    storage_name = f'sky-test-{int(time.time())}'
+    bucket_rclone_profile = Rclone.generate_rclone_bucket_profile_name(
+        storage_name, Rclone.RcloneClouds.IBM)
+    template_str = pathlib.Path(
+        'tests/test_yamls/test_ibm_cos_storage_mounting.yaml').read_text()
+    template = jinja2.Template(template_str)
+    content = template.render(storage_name=storage_name)
+    with tempfile.NamedTemporaryFile(suffix='.yaml', mode='w') as f:
+        f.write(content)
+        f.flush()
+        file_path = f.name
+        test_commands = [
+            *storage_setup_commands,
+            f'sky launch -y -c {name} --cloud ibm {file_path}',
+            f'sky logs {name} 1 --status',  # Ensure job succeeded.
+            f'rclone ls {bucket_rclone_profile}:{storage_name}/hello.txt',
+        ]
+        test = Test(
+            'ibm_storage_mounts',
+            test_commands,
+            f'sky down -y {name}; sky storage delete {storage_name}',
+            timeout=20 * 60,  # 20 mins
+        )
+        run_one_test(test)
+
+
 # ---------- CLI logs ----------
 @pytest.mark.no_scp  # SCP does not support num_nodes > 1 yet. Run test_scp_logs instead.
 def test_cli_logs(generic_cloud: str):
@@ -2446,6 +2477,24 @@ class TestStorageWithCredentials:
         'abc_',  # ends with an underscore
     ]
 
+    IBM_INVALID_NAMES = [
+        'ab',  # less than 3 characters
+        'abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz1',
+        # more than 63 characters
+        'Abcdef',  # contains an uppercase letter
+        'abc def',  # contains a space
+        'abc..def',  # two adjacent periods
+        '192.168.5.4',  # formatted as an IP address
+        'xn--bucket',  # starts with 'xn--' prefix
+        '.abc',  # starts with a dot
+        'abc.',  # ends with a dot
+        '-abc',  # starts with a hyphen
+        'abc-',  # ends with a hyphen
+        'a.-bc',  # contains the sequence '.-'
+        'a-.bc',  # contains the sequence '-.'
+        'a&bc'  # contains special characters
+        'ab^c'  # contains special characters
+    ]
     GITIGNORE_SYNC_TEST_DIR_STRUCTURE = {
         'double_asterisk': {
             'double_asterisk_excluded': None,
@@ -2550,6 +2599,10 @@ class TestStorageWithCredentials:
             endpoint_url = cloudflare.create_endpoint()
             url = f's3://{bucket_name}'
             return f'AWS_SHARED_CREDENTIALS_FILE={cloudflare.R2_CREDENTIALS_PATH} aws s3 rb {url} --force --endpoint {endpoint_url} --profile=r2'
+        if store_type == storage_lib.StoreType.IBM:
+            bucket_rclone_profile = Rclone.generate_rclone_bucket_profile_name(
+                bucket_name, Rclone.RcloneClouds.IBM)
+            return f'rclone purge {bucket_rclone_profile}:{bucket_name} && rclone config delete {bucket_rclone_profile}'
 
     @staticmethod
     def cli_ls_cmd(store_type, bucket_name, suffix=''):
@@ -2572,6 +2625,10 @@ class TestStorageWithCredentials:
             else:
                 url = f's3://{bucket_name}'
             return f'AWS_SHARED_CREDENTIALS_FILE={cloudflare.R2_CREDENTIALS_PATH} aws s3 ls {url} --endpoint {endpoint_url} --profile=r2'
+        if store_type == storage_lib.StoreType.IBM:
+            bucket_rclone_profile = Rclone.generate_rclone_bucket_profile_name(
+                bucket_name, Rclone.RcloneClouds.IBM)
+            return f'rclone ls {bucket_rclone_profile}:{bucket_name}/{suffix}'
 
     @staticmethod
     def cli_count_name_in_bucket(store_type, bucket_name, file_name, suffix=''):
@@ -2772,6 +2829,13 @@ class TestStorageWithCredentials:
             f'AWS_SHARED_CREDENTIALS_FILE={cloudflare.R2_CREDENTIALS_PATH} aws s3 rb s3://{tmp_bucket_name} --force --endpoint {endpoint_url} --profile=r2',
             shell=True)
 
+    @pytest.fixture
+    def tmp_ibm_cos_bucket(self, tmp_bucket_name):
+        # Creates a temporary bucket using IBM COS API
+        storage_obj = storage_lib.IBMCosStore(source="", name=tmp_bucket_name)
+        yield tmp_bucket_name
+        storage_obj.delete()
+
     @pytest.fixture
     def tmp_public_storage_obj(self, request):
         # Initializes a storage object with a public bucket
@@ -2782,6 +2846,7 @@ class TestStorageWithCredentials:
 
     @pytest.mark.parametrize('store_type', [
         storage_lib.StoreType.S3, storage_lib.StoreType.GCS,
+        pytest.param(storage_lib.StoreType.IBM, marks=pytest.mark.ibm),
         pytest.param(storage_lib.StoreType.R2, marks=pytest.mark.cloudflare)
     ])
     def test_new_bucket_creation_and_deletion(self, tmp_local_storage_obj,
@@ -2805,7 +2870,8 @@ class TestStorageWithCredentials:
     @pytest.mark.xdist_group('multiple_bucket_deletion')
     @pytest.mark.parametrize('store_type', [
         storage_lib.StoreType.S3, storage_lib.StoreType.GCS,
-        pytest.param(storage_lib.StoreType.R2, marks=pytest.mark.cloudflare)
+        pytest.param(storage_lib.StoreType.R2, marks=pytest.mark.cloudflare),
+        pytest.param(storage_lib.StoreType.IBM, marks=pytest.mark.ibm)
     ])
     def test_multiple_buckets_creation_and_deletion(
             self, tmp_multiple_scratch_storage_obj, store_type):
@@ -2843,6 +2909,7 @@ class TestStorageWithCredentials:
 
     @pytest.mark.parametrize('store_type', [
         storage_lib.StoreType.S3, storage_lib.StoreType.GCS,
+        pytest.param(storage_lib.StoreType.IBM, marks=pytest.mark.ibm),
         pytest.param(storage_lib.StoreType.R2, marks=pytest.mark.cloudflare)
     ])
     def test_bucket_external_deletion(self, tmp_scratch_storage_obj,
@@ -2871,6 +2938,7 @@ class TestStorageWithCredentials:
 
     @pytest.mark.parametrize('store_type', [
         storage_lib.StoreType.S3, storage_lib.StoreType.GCS,
+        pytest.param(storage_lib.StoreType.IBM, marks=pytest.mark.ibm),
         pytest.param(storage_lib.StoreType.R2, marks=pytest.mark.cloudflare)
     ])
     def test_bucket_bulk_deletion(self, store_type, tmp_bulk_del_storage_obj):
@@ -2901,6 +2969,7 @@ class TestStorageWithCredentials:
 
     @pytest.mark.parametrize('nonexist_bucket_url', [
         's3://{random_name}', 'gs://{random_name}',
+        pytest.param('cos://us-east/{random_name}', marks=pytest.mark.ibm),
         pytest.param('r2://{random_name}', marks=pytest.mark.cloudflare)
     ])
     def test_nonexistent_bucket(self, nonexist_bucket_url):
@@ -2919,6 +2988,19 @@ class TestStorageWithCredentials:
                 endpoint_url = cloudflare.create_endpoint()
                 command = f'AWS_SHARED_CREDENTIALS_FILE={cloudflare.R2_CREDENTIALS_PATH} aws s3api head-bucket --bucket {nonexist_bucket_name} --endpoint {endpoint_url} --profile=r2'
                 expected_output = '404'
+            elif nonexist_bucket_url.startswith('cos'):
+                # Using API calls, since using rclone requires a profile's name
+                try:
+                    expected_output = command = "echo"  # avoid unrelated exception in case of failure.
+                    bucket_name = urllib.parse.urlsplit(
+                        nonexist_bucket_url.format(
+                            random_name=nonexist_bucket_name)).path.strip('/')
+                    client = ibm.get_cos_client('us-east')
+                    client.head_bucket(Bucket=bucket_name)
+                except ibm.ibm_botocore.exceptions.ClientError as e:
+                    if e.response['Error']['Code'] == '404':
+                        # success
+                        return
             else:
                 raise ValueError('Unsupported bucket type '
                                  f'{nonexist_bucket_url}')
@@ -2946,13 +3028,17 @@ class TestStorageWithCredentials:
             storage_obj = storage_lib.Storage(source=nonexist_bucket_url.format(
                 random_name=nonexist_bucket_name))
 
-    @pytest.mark.parametrize('private_bucket',
-                             [f's3://imagenet', f'gs://imagenet'])
+    @pytest.mark.parametrize('private_bucket', [
+        f's3://imagenet', f'gs://imagenet',
+        pytest.param('cos://us-east/bucket1', marks=pytest.mark.ibm)
+    ])
     def test_private_bucket(self, private_bucket):
         # Attempts to access private buckets not belonging to the user.
         # These buckets are known to be private, but may need to be updated if
         # they are removed by their owners.
-        private_bucket_name = urllib.parse.urlsplit(private_bucket).netloc
+        private_bucket_name = urllib.parse.urlsplit(private_bucket).netloc if \
+              urllib.parse.urlsplit(private_bucket).scheme != 'cos' else \
+                  urllib.parse.urlsplit(private_bucket).path.strip('/')
         with pytest.raises(
                 sky.exceptions.StorageBucketGetError,
                 match=storage_lib._BUCKET_FAIL_TO_CONNECT_MESSAGE.format(
@@ -2962,6 +3048,9 @@ class TestStorageWithCredentials:
     @pytest.mark.parametrize('ext_bucket_fixture, store_type',
                              [('tmp_awscli_bucket', storage_lib.StoreType.S3),
                               ('tmp_gsutil_bucket', storage_lib.StoreType.GCS),
+                              pytest.param('tmp_ibm_cos_bucket',
+                                           storage_lib.StoreType.IBM,
+                                           marks=pytest.mark.ibm),
                               pytest.param('tmp_awscli_bucket_r2',
                                            storage_lib.StoreType.R2,
                                            marks=pytest.mark.cloudflare)])
@@ -3006,6 +3095,7 @@ class TestStorageWithCredentials:
 
     @pytest.mark.parametrize('store_type', [
         storage_lib.StoreType.S3, storage_lib.StoreType.GCS,
+        pytest.param(storage_lib.StoreType.IBM, marks=pytest.mark.ibm),
         pytest.param(storage_lib.StoreType.R2, marks=pytest.mark.cloudflare)
     ])
     def test_list_source(self, tmp_local_list_storage_obj, store_type):
@@ -3032,6 +3122,9 @@ class TestStorageWithCredentials:
     @pytest.mark.parametrize('invalid_name_list, store_type',
                              [(AWS_INVALID_NAMES, storage_lib.StoreType.S3),
                               (GCS_INVALID_NAMES, storage_lib.StoreType.GCS),
+                              pytest.param(IBM_INVALID_NAMES,
+                                           storage_lib.StoreType.IBM,
+                                           marks=pytest.mark.ibm),
                               pytest.param(AWS_INVALID_NAMES,
                                            storage_lib.StoreType.R2,
                                            marks=pytest.mark.cloudflare)])
diff --git a/tests/test_yamls/test_ibm_cos_storage_mounting.yaml b/tests/test_yamls/test_ibm_cos_storage_mounting.yaml
new file mode 100644
index 000000000..cbfcd045f
--- /dev/null
+++ b/tests/test_yamls/test_ibm_cos_storage_mounting.yaml
@@ -0,0 +1,40 @@
+file_mounts:
+  # Mounting private buckets in COPY mode with a source dir
+  /mount_private_copy:
+    name: {{storage_name}}
+    source: ~/tmp-workdir
+    store: ibm
+    mode: COPY
+
+  # Mounting private buckets in COPY mode with a list of files as source
+  /mount_private_copy_lof:
+    name: {{storage_name}}
+    source: ['~/tmp-workdir/tmp file', '~/tmp-workdir/tmp file2']
+    store: ibm
+    mode: COPY
+
+  # Mounting private buckets in MOUNT mode
+  /mount_private_mount:
+    name: {{storage_name}}
+    source: ~/tmp-workdir
+    store: ibm
+    mode: MOUNT
+
+run: |
+  set -ex
+
+  # Check private bucket contents
+  ls -ltr /mount_private_copy/foo
+  ls -ltr /mount_private_copy/tmp\ file
+  ls -ltr /mount_private_copy_lof/tmp\ file
+  ls -ltr /mount_private_copy_lof/tmp\ file2
+  ls -ltr /mount_private_mount/foo
+  ls -ltr /mount_private_mount/tmp\ file
+  
+  # Symlinks are not copied to buckets
+  ! ls /mount_private_copy/circle-link
+  ! ls /mount_private_mount/circle-link
+  
+  # Write to private bucket in MOUNT mode should pass
+  echo "hello" > /mount_private_mount/hello.txt
+