Repository: skypilot-org/skypilot

Issue:
[Spot] Failed spot jobs do not have complete log in the `sky spot logs --controller`
The current implementation tries to stream the logs for the failed spot jobs back to controller after the user program fails. However,  if the output is too long, the streaming will be interrupted in the middle, causing the error log not complete in the `sky spot logs --controller`.

This has been observed by many users.

To reproduce:
`sky spot launch task.yaml`

```
# task.yaml
run: |
  for i in `seq 1 1000`; do
    echo very long outputs
  done
  mkdir -p $HOME/data
  wget https://huggingface.co/datasets/anon8231489123/ShareGPT_Vicuna_unfiltered/resolve/main/ShareGPT_V3_unfiltered_cleaned_split.json  -O $HOME/data/sharegpt.json
  echo failed run
  exit 1
```


The output of `sky spot logs --controller` will be:
```
...
....... .......... .......... .......... 12%  104M 2s
(failed-run, pid=14930) (failed-run, pid=3781)  83350K .......... .......... .......... .......... .......... 12%  386M 2s
(failed-run, pid=14930) (failed-run, pid=3781)  83400K .......... .......... .......... .......... .......... 12%  401M 2s
(failed-run, pid=14930) (failed-run, pid=3781)  83450K .......... .......... .......... .......... .......... 12%  348M 2s
(failed-run, pid=14930) (failed-run, pid=3781)  83500K .......... .......... .......... .......... .......... 12%  376M 2s
(failed-run, pid=14930) (failed-run, pid=3781)  83550K .......... .........: Resource temporarily unavailable
(failed-run, pid=14930) Shared connection to 34.130.221.137 closed.
```

Discussion:

Patch:
diff --git a/docs/source/getting-started/installation.rst b/docs/source/getting-started/installation.rst
index 2fc23110..71431222 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 b8ee4393..dcf9958b 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 0395a482..9cffdca8 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 502dec40..5303a538 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/cloud_stores.py b/sky/cloud_stores.py
index f88d3c13..b0535603 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:
@@ -223,8 +224,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 a4e4f275..b666fdc8 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 eeb827df..0882f755 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 66276e0f..9634ce77 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 562910d9..c9ed89f0 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 7a5915ab..b2ace9a7 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 4ecec9eb..7bd68f5a 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 7e8fb37b..39c2727c 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 81a43cb4..cbb8a621 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 cb593247..050e33b2 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 00000000..cbfcd045
--- /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
+