Repository: skypilot-org/skypilot
Pull request #1598 (2023-01-16 14:23:05)

IBM cloud provider integration 
<!-- Describe the changes in this PR -->
PR Topic - integration of IBM as a new cloud provider. 
Successfully ran the following tests: 
- test_ibm_n_node_job_queue
- test_cancel_ibm
- test_ibm_autodown
- test_ibm_autostop
- test_ibm_job_queue
- test_ibm_file_mounts
- test_ibm_zone
- test_ibm_region
- test_ibm_minimal  

Notes: 
- Had to introduce new tests (that slightly differ from their original counterparts), since current tests are tailored for specific cloud providers. 
- Couldn't run tests as a batch, since some pytest config files are requiring aws credentials to run.  
- Couldn't run backwards compatibility test for reason mentioned above.  

I plan to:
 - Create a script users could run to generate a credential file located in `~/.ibm`, similarly to `aws configure`.
 - Add support for IBM's cloud object storage.  

Setup: 
Until the above mentioned script will be created, authenticate your IBM account by adding the following data in `~/.ibm/credentials.yaml` :
```
iam_api_key: <String>
resource_group_id: <String>
```
- IBM cloud API key can be created on the IBM Cloud console: Manage > Access (IAM) > Create an IBM Cloud API key. 
For the complete guide visit [this](https://www.ibm.com/docs/en/mas-cd?topic=cli-creating-your-cloud-api-key) link.
- Resource group id can be found in the IBM Cloud console: Manage > Account > Account resources > Resource groups. For the complete guide visit [this](https://cloud.ibm.com/docs/account?topic=account-rgs&interface=ui) link.



ewzeng: Thanks for your hard work @Cohen-J-Omer!

Are there any instructions for us to try out this PR (e.g. create an account on IBM cloud, save IBM credentials to a file, etc.)?
Cohen-J-Omer: > Thanks for your hard work @Cohen-J-Omer!
> 
> Are there any instructions for us to try out this PR (e.g. create an account on IBM cloud, save IBM credentials to a certain file, etc.)?

Hey @ewzeng, 
You're right. I've added a setup segment to the PR's description. 
Once the PR will get approved I'll:
- Create a script to automate the process, including configuration options, such as base image selection.
- Add an IBM setup segment to Readme.md as a separate PR (or include in this PR if you'd prefer).     
gilv: @ewzeng great you assigned to this PR... Do you plan to test it with IBM Cloud account? Do you have one? The account need to have permissions to setup VPc..
ewzeng: @Cohen-J-Omer Thanks! I agree -- we can update the docs in a separate PR.

@gilv Yes I was planning to make an IBM cloud account to test it out (I don't have one yet). Is there anything special I should do to get the necessary account permissions?
gilv: @ewzeng just regular account will do the trick... but if you use all kind of academic accounts or free trials, then I think VPC is blocked there
ewzeng: @gilv Okay, got it. This is good to know. Thanks!
ewzeng: @Cohen-J-Omer Here are some comments from trying out this pr! (I will go through the code soon)

1. If IBM credentials are not set up, `sky check` returns
```
/home/zeng/.ibm/credentials.yaml does not exist. Run the following command:
a configuration script that asks for
api key and stores it in ~/.ibm/credentials.yaml
```
Let's replace this with actual instructions for now (and add a TODO statement in the code to make a configuration script)

2. In `sky launch`, it seems that we `ssh` into the node (and run set up commands) as the root user. We should probably use the `ubuntu` user instead.

3. After `sky launch --cloud ibm -i 1 --down`, it seems that the vpc and security group don't get deleted.

4. This might be a bug in IBM's UI, but when I `sky stop` a two-node cluster, the UI says that the worker node is `Stopped` while the head node is `Stopping` (no matter how long I wait or refresh the page).

5. `sky launch --cloud ibm examples/huggingface_glue_imdb_app.yaml` results in a `sqlite` error on the launched node. (Perhaps this has something to do with 2)

6. It seems that `nvidia-smi` is not automatically installed on IBM nodes; since some of the smoke tests use `nvidia-smi`, we should consider installing it.

Overall, awesome work! I tested `sky start/stop/launch/down/status/exec` along with autodown, autostop, and multi-node, and the experience felt very smooth.
ewzeng: Thanks for the updates @Cohen-J-Omer! I will test this out.
ewzeng: @Cohen-J-Omer I tried running
```
sky launch --cloud ibm   # launches cluster sky-f6c7-zeng
sky autostop sky-f6c7-zeng -i 2 --down
```
and waited for about 10 minutes, but the cluster was still up. Are you able to replicate this?
Cohen-J-Omer: @ewzeng thanks for noticing! 
Ray's cli_logger caused an unclear exception about the logging messages i fed it. Replaced it and made the implementation a bit more robust. Please re-pull.    
ewzeng: I re-pulled! But now when I run
```
sky launch --cloud ibm --down
```
the instance still stays up (waited for ~10 minutes).
Cohen-J-Omer: (1) I could add it until a script is made, since it will locate the resource group id and will suggest applying optional fields.    
(3) It works on my end. I'd like to get more info on slack.
(6) yes, the same one. it will be an interactive tool that will ask for an `IAM-API-key` and according to user's choices will generate a credential file.  
 
 
Cohen-J-Omer: Hey @ewzeng, 
I've rebased, please review the changes. 
Cohen-J-Omer: Hey @Michaelvll,
I've updated the modules as requested.
The following tests were executed successfully:
- test_ibm_region
- test_ibm_zone
- test_ibm_cancel_task
- test_ibm_job_queue
- test_ibm_job_queue_multinode
- test_file_mounts
- test_minimal
- test_autodown

- The `stop` functionality remains to be problematic on skypilot, e.g. test_autostop sporadically succeeds. A worker node is indefinitely stuck on `uninitialized` status.
- Would add an `image_id` test, as the functionality is implemented, but I cannot publish a VPC image to the IBM public VPC image catalog, which I could then rely on its constant ID (for testing purposes). 

If the PR is in a satisfactory condition I will rebase it (or merge if the process will prove too difficult due to numerous conflicts). 
Cohen-J-Omer: Hey @Michaelvll,
- I've made the required changes.
- `format.sh` passes all checks (Pylint, mypy, Black, yapf). 
- I've marked tests that cannot be executed successfully on IBM cloud. Documented explanations.
- I can't execute test `test_example_app`, as it requires GCP credentials (only have access to IBM and AWS), so I currently marked it `@pytest.mark.no_ibm` to avoid potential incompatibility. 
- Since each cluster is created on a different VPC, while considering the default [limit](https://cloud.ibm.com/docs/vpc?topic=vpc-quotas#vpc-quotas) of 10 VPCs per region, only up to 10 tests can be executed simultaneously. To increase said limit open a [case](https://cloud.ibm.com/unifiedsupport/cases/form) with IBM support.  

If the PR is in a satisfactory condition, let me know and i'll rebase/merge. 
Cohen-J-Omer: Hey @Michaelvll,
I've solved conflicts via a rebase. Please review. 

Patch:
diff --git a/.github/workflows/format.yml b/.github/workflows/format.yml
index 05b85431b..19783e641 100644
--- a/.github/workflows/format.yml
+++ b/.github/workflows/format.yml
@@ -36,9 +36,11 @@ jobs:
         yapf --diff --recursive ./ --exclude 'sky/skylet/ray_patches/**' \
             --exclude 'sky/skylet/providers/aws/**' \
             --exclude 'sky/skylet/providers/gcp/**' \
-            --exclude 'sky/skylet/providers/azure/**'
+            --exclude 'sky/skylet/providers/azure/**' \
+            --exclude 'sky/skylet/providers/ibm/**'
     - name: Running black
       run: |
         black --diff --check sky/skylet/providers/aws/ \
             sky/skylet/providers/gcp/ \
-            sky/skylet/providers/azure/
+            sky/skylet/providers/azure/ \
+            sky/skylet/providers/ibm/
diff --git a/docs/source/examples/spot-jobs.rst b/docs/source/examples/spot-jobs.rst
index 2ddd47a1f..68c5316b8 100644
--- a/docs/source/examples/spot-jobs.rst
+++ b/docs/source/examples/spot-jobs.rst
@@ -223,7 +223,7 @@ Cancel a spot job:
 Real-world examples
 -------------------------
 
-* `Vicuna <https://vicuna.lmsys.org/>`_ LLM chatbot: `instructions <https://github.com/skypilot-org/skypilot/tree/master/examples/vicuna-llm>`_, `YAML <https://github.com/lm-sys/FastChat/blob/main/scripts/train-alpaca.yaml>`_
+* `Vicuna <https://vicuna.lmsys.org/>`_ LLM chatbot: `instructions <https://github.com/skypilot-org/skypilot/tree/master/llm/vicuna>`_, `YAML <https://github.com/skypilot-org/skypilot/blob/master/llm/vicuna/train.yaml>`_
 * BERT (shown above): `YAML <https://github.com/skypilot-org/skypilot/blob/master/examples/spot/bert_qa.yaml>`_
 * PyTorch DDP, ResNet: `YAML <https://github.com/skypilot-org/skypilot/blob/master/examples/spot/resnet.yaml>`_
 * PyTorch Lightning DDP, CIFAR-10: `YAML <https://github.com/skypilot-org/skypilot/blob/master/examples/spot/lightning_cifar10.yaml>`_
diff --git a/docs/source/reference/storage.rst b/docs/source/reference/storage.rst
index 46703a25b..b8ee43939 100644
--- a/docs/source/reference/storage.rst
+++ b/docs/source/reference/storage.rst
@@ -151,7 +151,7 @@ and storage mounting:
       # this approach can also be used to create a shared filesystem across workers.
       # See examples/storage/pingpong.yaml for an example.
       /outputs-mount:
-        name: romil-output-bucket
+        name: my-output-bucket
         mode: MOUNT
 
       # *** Uploading multiple files to the same Storage object ***
diff --git a/docs/source/reference/yaml-spec.rst b/docs/source/reference/yaml-spec.rst
index a4bccf66e..66d264189 100644
--- a/docs/source/reference/yaml-spec.rst
+++ b/docs/source/reference/yaml-spec.rst
@@ -121,6 +121,16 @@ Available fields:
       # GCP
       # To find GCP images: https://cloud.google.com/compute/docs/images
       # image_id: projects/deeplearning-platform-release/global/images/family/tf2-ent-2-1-cpu-ubuntu-2004
+      #
+      # IBM
+      # Create a private VPC image and paste its ID in the following format:
+      # image_id: <unique_image_id>
+      # To create an image manually:
+      # https://cloud.ibm.com/docs/vpc?topic=vpc-creating-and-using-an-image-from-volume.
+      # To use an official VPC image creation tool:
+      # https://www.ibm.com/cloud/blog/use-ibm-packer-plugin-to-create-custom-images-on-ibm-cloud-vpc-infrastructure
+      # To use a more limited but easier to manage tool:
+      # https://github.com/IBM/vpc-img-inst
 
     file_mounts:
       # Uses rsync to sync local files/directories to all nodes of the cluster.
diff --git a/examples/job_queue/job_ibm.yaml b/examples/job_queue/job_ibm.yaml
new file mode 100644
index 000000000..62bdcfd26
--- /dev/null
+++ b/examples/job_queue/job_ibm.yaml
@@ -0,0 +1,23 @@
+# A task submitted to an existing cluster.
+#
+# Runs a task on a existing cluster.
+#
+# Usage:
+#   sky launch -c jq cluster.yaml
+#   sky exec jq job_ibm.yaml
+
+name: job
+
+resources:
+  accelerators: v100:0.5
+
+setup: |
+  echo "running setup"
+
+run: |
+  timestamp=$(date +%s)
+  conda env list
+  for i in {1..120}; do
+    echo "$timestamp $i"
+    sleep 1
+  done
diff --git a/examples/job_queue/job_multinode_ibm.yaml b/examples/job_queue/job_multinode_ibm.yaml
new file mode 100644
index 000000000..cce8edcfd
--- /dev/null
+++ b/examples/job_queue/job_multinode_ibm.yaml
@@ -0,0 +1,27 @@
+# A task runs on an existing multinode cluster.
+#
+# Runs a task that requires multinode.
+#
+# Usage:
+#   sky launch -c mjq cluster_multinode.yaml
+#   sky exec mjq job_multinode_ibm.yaml
+#   sky exec mjq job_ibm.yaml
+
+name: job_multinode
+
+resources:
+  accelerators: v100:0.5
+
+num_nodes: 2
+
+setup: |
+  echo "running setup"
+  sleep 80
+
+run: |
+  timestamp=$(date +%s)
+  conda env list
+  for i in {1..240}; do
+    echo "$timestamp $i"
+    sleep 1
+  done
diff --git a/format.sh b/format.sh
index 281a51f55..8857fe759 100755
--- a/format.sh
+++ b/format.sh
@@ -50,12 +50,14 @@ YAPF_EXCLUDES=(
     '--exclude' 'sky/skylet/providers/aws/**'
     '--exclude' 'sky/skylet/providers/gcp/**'
     '--exclude' 'sky/skylet/providers/azure/**'
+    '--exclude' 'sky/skylet/providers/ibm/**'
 )
 
 BLACK_INCLUDES=(
     'sky/skylet/providers/aws'
     'sky/skylet/providers/gcp'
     'sky/skylet/providers/azure'
+    'sky/skylet/providers/ibm'
 )
 
 # Format specified files
diff --git a/sky/__init__.py b/sky/__init__.py
index 3462fd67d..41e5f3d24 100644
--- a/sky/__init__.py
+++ b/sky/__init__.py
@@ -25,6 +25,7 @@ from sky.core import (status, start, stop, down, autostop, queue, cancel,
                       cost_report)
 
 # Aliases.
+IBM = clouds.IBM
 AWS = clouds.AWS
 Azure = clouds.Azure
 GCP = clouds.GCP
@@ -34,6 +35,7 @@ optimize = Optimizer.optimize
 
 __all__ = [
     '__version__',
+    'IBM',
     'AWS',
     'Azure',
     'GCP',
diff --git a/sky/adaptors/ibm.py b/sky/adaptors/ibm.py
new file mode 100644
index 000000000..0395a4820
--- /dev/null
+++ b/sky/adaptors/ibm.py
@@ -0,0 +1,107 @@
+"""IBM cloud adaptors"""
+
+# pylint: disable=import-outside-toplevel
+
+from sky import sky_logging
+import yaml
+import os
+import json
+import requests
+import functools
+
+CREDENTIAL_FILE = '~/.ibm/credentials.yaml'
+logger = sky_logging.init_logger(__name__)
+
+ibm_vpc = None
+ibm_cloud_sdk_core = None
+ibm_platform_services = None
+
+
+def import_package(func):
+
+    @functools.wraps(func)
+    def wrapper(*args, **kwargs):
+        global ibm_vpc, ibm_cloud_sdk_core, ibm_platform_services
+        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
+                ibm_vpc = _ibm_vpc
+                ibm_cloud_sdk_core = _ibm_cloud_sdk_core
+                ibm_platform_services = _ibm_platform_services
+            except ImportError:
+                raise ImportError(
+                    'Failed to import dependencies for IBM. '
+                    'Try running: pip install "skypilot[ibm]".\n') from None
+        return func(*args, **kwargs)
+
+    return wrapper
+
+
+def read_credential_file():
+    with open(os.path.expanduser(CREDENTIAL_FILE), encoding='utf-8') as f:
+        return yaml.safe_load(f)
+
+
+def get_api_key():
+    return read_credential_file()['iam_api_key']
+
+
+@import_package
+def _get_authenticator():
+    return ibm_cloud_sdk_core.authenticators.IAMAuthenticator(get_api_key())
+
+
+def get_oauth_token():
+    """returns a temporary authentication token required by
+        various IBM cloud APIs
+    using an http request to avoid having to install module
+        ibm_watson to get IAMTokenManager"""
+    # pylint: disable=line-too-long
+    res = requests.post(
+        'https://iam.cloud.ibm.com/identity/token',
+        headers={'Content-Type': 'application/x-www-form-urlencoded'},
+        data=
+        f'grant_type=urn:ibm:params:oauth:grant-type:apikey&apikey={get_api_key()}'
+    )
+    return json.loads(res.text)['access_token']
+
+
+@import_package
+def client(**kwargs):
+    """Create an ibm vpc client.
+
+    Sets the vpc client to a specific region.
+    If none was specified 'us-south' is set internally.
+
+    Args:
+        kwargs: Keyword arguments.
+
+    Returns:
+        ibm vpc client
+    """
+
+    try:
+        vpc_client = ibm_vpc.VpcV1(version='2022-06-30',
+                                   authenticator=_get_authenticator())
+        if kwargs.get('region'):
+            vpc_client.set_service_url(
+                f'https://{kwargs["region"]}.iaas.cloud.ibm.com/v1')
+    except Exception:
+        logger.error('No registered API key found matching specified value')
+        raise
+
+    return vpc_client  # returns either formerly or newly created client
+
+
+@import_package
+def search_client():
+    return ibm_platform_services.GlobalSearchV2(
+        authenticator=_get_authenticator())
+
+
+@import_package
+def tagging_client():
+    return ibm_platform_services.GlobalTaggingV1(
+        authenticator=_get_authenticator())
diff --git a/sky/authentication.py b/sky/authentication.py
index fa93e77f7..91d904440 100644
--- a/sky/authentication.py
+++ b/sky/authentication.py
@@ -9,6 +9,7 @@ import sys
 import textwrap
 import time
 from typing import Any, Dict, Tuple
+import uuid
 
 import colorama
 from cryptography.hazmat.primitives import serialization
@@ -17,7 +18,7 @@ from cryptography.hazmat.backends import default_backend
 
 from sky import clouds
 from sky import sky_logging
-from sky.adaptors import gcp
+from sky.adaptors import gcp, ibm
 from sky.utils import common_utils
 from sky.utils import subprocess_utils
 from sky.utils import ux_utils
@@ -334,3 +335,61 @@ def setup_lambda_authentication(config: Dict[str, Any]) -> Dict[str, Any]:
     config['file_mounts'] = file_mounts
 
     return config
+
+
+def setup_ibm_authentication(config):
+    """ registers keys if they do not exist in sky folder
+    and updates config file.
+    keys default location: '~/.ssh/sky-key' and '~/.ssh/sky-key.pub'
+    """
+
+    def _get_unique_key_name():
+        suffix_len = 10
+        return f'skypilot-key-{str(uuid.uuid4())[:suffix_len]}'
+
+    client = ibm.client(region=config['provider']['region'])
+    resource_group_id = config['provider']['resource_group_id']
+
+    _, public_key_path = get_or_generate_keys()
+    with open(os.path.abspath(os.path.expanduser(public_key_path)),
+              'r',
+              encoding='utf-8') as file:
+        ssh_key_data = file.read()
+    # pylint: disable=E1136
+    try:
+        res = client.create_key(public_key=ssh_key_data,
+                                name=_get_unique_key_name(),
+                                resource_group={
+                                    'id': resource_group_id
+                                },
+                                type='rsa').get_result()
+        vpc_key_id = res['id']
+        logger.debug(f'Created new key: {res["name"]}')
+
+    except ibm.ibm_cloud_sdk_core.ApiException as e:
+        if 'Key with fingerprint already exists' in e.message:
+            for key in client.list_keys().result['keys']:
+                if (ssh_key_data in key['public_key'] or
+                        key['public_key'] in ssh_key_data):
+                    vpc_key_id = key['id']
+                    logger.debug(f'Reusing key:{key["name"]}, '
+                                 f'matching existing public key.')
+                    break
+        elif 'Key with name already exists' in e.message:
+            raise Exception("""a key with chosen name
+                already registered in the specified region""") from e
+        else:
+            raise Exception('Failed to register a key') from e
+
+    config['auth']['ssh_private_key'] = PRIVATE_SSH_KEY_PATH
+
+    for node_type in config['available_node_types']:
+        config['available_node_types'][node_type]['node_config'][
+            'key_id'] = vpc_key_id
+
+    # Add public key path to file mounts
+    file_mounts = config['file_mounts']
+    file_mounts[PUBLIC_SSH_KEY_PATH] = PUBLIC_SSH_KEY_PATH
+    config['file_mounts'] = file_mounts
+
+    return config
diff --git a/sky/backends/backend_utils.py b/sky/backends/backend_utils.py
index df2d65a0f..3bf61f1f3 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -51,6 +51,7 @@ from sky.utils import tpu_utils
 from sky.utils import ux_utils
 from sky.utils import validator
 from sky.usage import usage_lib
+from sky.adaptors import ibm
 
 if typing.TYPE_CHECKING:
     from sky import resources
@@ -991,6 +992,8 @@ def _add_auth_to_cluster_config(cloud: clouds.Cloud, cluster_config_file: str):
         config = auth.setup_azure_authentication(config)
     elif isinstance(cloud, clouds.Lambda):
         config = auth.setup_lambda_authentication(config)
+    elif isinstance(cloud, clouds.IBM):
+        config = auth.setup_ibm_authentication(config)
     else:
         assert isinstance(cloud, clouds.Local), cloud
         # Local cluster case, authentication is already filled by the user
@@ -1702,6 +1705,48 @@ def _query_status_azure(
     return _process_cli_query('Azure', cluster, query_cmd, '\t', status_map)
 
 
+def _query_status_ibm(
+    cluster: str,
+    ray_config: Dict[str, Any],
+) -> List[global_user_state.ClusterStatus]:
+    """
+    returns a list of Statuses for each of the cluster's nodes.
+    this function gets called when running `sky status` with -r flag and the cluster's head node is either stopped or down.
+    """
+
+    status_map: Dict[str, Any] = {
+        'pending': global_user_state.ClusterStatus.INIT,
+        'starting': global_user_state.ClusterStatus.INIT,
+        'restarting': global_user_state.ClusterStatus.INIT,
+        'running': global_user_state.ClusterStatus.UP,
+        'stopping': global_user_state.ClusterStatus.STOPPED,
+        'stopped': global_user_state.ClusterStatus.STOPPED,
+        'deleting': None,
+        'failed': global_user_state.ClusterStatus.INIT,
+        'cluster_deleted': []
+    }
+
+    client = ibm.client(region=ray_config['provider']['region'])
+    search_client = ibm.search_client()
+    # pylint: disable=E1136
+    vpcs_filtered_by_tags_and_region = search_client.search(
+        query=
+        f'type:vpc AND tags:{cluster} AND region:{ray_config["provider"]["region"]}',
+        fields=['tags', 'region', 'type'],
+        limit=1000).get_result()['items']
+    if not vpcs_filtered_by_tags_and_region:
+        # a vpc could have been removed unkownlingly to skypilot, such as
+        # via `sky autostop --down`, or simply manually (e.g. via console).
+        logger.warning('No vpc exists in '
+                       f'{ray_config["provider"]["region"]} '
+                       f'with tag: {cluster}')
+        return status_map['cluster_deleted']
+    vpc_id = vpcs_filtered_by_tags_and_region[0]['crn'].rsplit(':', 1)[-1]
+    instances = client.list_instances(vpc_id=vpc_id).get_result()['instances']
+
+    return [status_map[instance['status']] for instance in instances]
+
+
 def _query_status_lambda(
         cluster: str,
         ray_config: Dict[str, Any],  # pylint: disable=unused-argument
@@ -1729,6 +1774,7 @@ _QUERY_STATUS_FUNCS = {
     'GCP': _query_status_gcp,
     'Azure': _query_status_azure,
     'Lambda': _query_status_lambda,
+    'IBM': _query_status_ibm,
 }
 
 
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index 67163a78f..a8cbf9164 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -70,6 +70,7 @@ _NODES_LAUNCHING_PROGRESS_TIMEOUT = {
     clouds.Azure: 90,
     clouds.GCP: 120,
     clouds.Lambda: 150,
+    clouds.IBM: 160,
     clouds.Local: 90,
 }
 
@@ -120,6 +121,7 @@ def _get_cluster_config_template(cloud):
         clouds.Azure: 'azure-ray.yml.j2',
         clouds.GCP: 'gcp-ray.yml.j2',
         clouds.Lambda: 'lambda-ray.yml.j2',
+        clouds.IBM: 'ibm-ray.yml.j2',
         clouds.Local: 'local-ray.yml.j2',
     }
     return cloud_to_template[type(cloud)]
@@ -842,6 +844,37 @@ class RetryingVmProvisioner(object):
                         self._blocked_resources.add(
                             launchable_resources.copy(region=r.name, zone=None))
 
+    def _update_blocklist_on_ibm_error(
+            self, launchable_resources: 'resources_lib.Resources',
+            region: 'clouds.Region', zones: Optional[List['clouds.Zone']],
+            stdout: str, stderr: str):
+
+        style = colorama.Style
+        stdout_splits = stdout.split('\n')
+        stderr_splits = stderr.split('\n')
+        errors = [
+            s.strip()
+            for s in stdout_splits + stderr_splits
+            if 'ERR' in s.strip() or 'PANIC' in s.strip()
+        ]
+        if not errors:
+            logger.info('====== stdout ======')
+            for s in stdout_splits:
+                print(s)
+            logger.info('====== stderr ======')
+            for s in stderr_splits:
+                print(s)
+            with ux_utils.print_exception_no_traceback():
+                raise RuntimeError('Errors occurred during provision; '
+                                   'check logs above.')
+        logger.warning(f'Got error(s) on IBM cluster, in {region.name}:')
+        messages = '\n\t'.join(errors)
+        logger.warning(f'{style.DIM}\t{messages}{style.RESET_ALL}')
+
+        for zone in zones:  # type: ignore[union-attr]
+            self._blocked_resources.add(
+                launchable_resources.copy(zone=zone.name))
+
     def _update_blocklist_on_local_error(
             self, launchable_resources: 'resources_lib.Resources',
             region: 'clouds.Region', zones: Optional[List['clouds.Zone']],
@@ -906,6 +939,7 @@ class RetryingVmProvisioner(object):
             clouds.Azure: self._update_blocklist_on_azure_error,
             clouds.GCP: self._update_blocklist_on_gcp_error,
             clouds.Lambda: self._update_blocklist_on_lambda_error,
+            clouds.IBM: self._update_blocklist_on_ibm_error,
             clouds.Local: self._update_blocklist_on_local_error,
         }
         cloud = launchable_resources.cloud
@@ -3041,6 +3075,42 @@ class CloudVmRayBackend(backends.Backend['CloudVmRayResourceHandle']):
                     shell=True,
                     stream_logs=False,
                     require_outputs=True)
+
+        elif (isinstance(cloud, clouds.IBM) and terminate and
+              prev_cluster_status == global_user_state.ClusterStatus.STOPPED):
+            # pylint: disable= W0622 W0703 C0415
+            from sky.adaptors import ibm
+            from sky.skylet.providers.ibm.vpc_provider import IBMVPCProvider
+
+            config_provider = common_utils.read_yaml(
+                handle.cluster_yaml)['provider']
+            region = config_provider['region']
+            cluster_name = handle.cluster_name
+            search_client = ibm.search_client()
+            vpc_found = False
+            # pylint: disable=unsubscriptable-object
+            vpcs_filtered_by_tags_and_region = search_client.search(
+                query=f'type:vpc AND tags:{cluster_name} AND region:{region}',
+                fields=['tags', 'region', 'type'],
+                limit=1000).get_result()['items']
+            try:
+                # pylint: disable=line-too-long
+                vpc_id = vpcs_filtered_by_tags_and_region[0]['crn'].rsplit(
+                    ':', 1)[-1]
+                vpc_found = True
+            except Exception:
+                logger.critical('failed to locate vpc for ibm cloud')
+                returncode = -1
+
+            if vpc_found:
+                # # pylint: disable=line-too-long E1136
+                # Delete VPC and it's associated resources
+                vpc_provider = IBMVPCProvider(
+                    config_provider['resource_group_id'], region, cluster_name)
+                vpc_provider.delete_vpc(vpc_id, region)
+                # successfully removed cluster as no exception was raised
+                returncode = 0
+
         elif (terminate and
               (prev_cluster_status == global_user_state.ClusterStatus.STOPPED or
                use_tpu_vm)):
diff --git a/sky/clouds/__init__.py b/sky/clouds/__init__.py
index ca442dcb5..e2c379376 100644
--- a/sky/clouds/__init__.py
+++ b/sky/clouds/__init__.py
@@ -9,8 +9,10 @@ from sky.clouds.azure import Azure
 from sky.clouds.gcp import GCP
 from sky.clouds.lambda_cloud import Lambda
 from sky.clouds.local import Local
+from sky.clouds.ibm import IBM
 
 __all__ = [
+    'IBM',
     'AWS',
     'Azure',
     'Cloud',
diff --git a/sky/clouds/ibm.py b/sky/clouds/ibm.py
new file mode 100644
index 000000000..f1733a4e1
--- /dev/null
+++ b/sky/clouds/ibm.py
@@ -0,0 +1,443 @@
+"""IBM Web Services."""
+import os
+import yaml
+import json
+from sky import clouds
+from sky.clouds import service_catalog
+from sky.adaptors import ibm
+from sky import sky_logging
+import typing
+from typing import Dict, Iterator, List, Optional, Tuple
+from sky.adaptors.ibm import CREDENTIAL_FILE
+
+from sky.clouds.cloud import Zone
+from sky.utils import ux_utils
+if typing.TYPE_CHECKING:
+    # renaming to avoid shadowing variables
+    from sky import resources as resources_lib
+
+logger = sky_logging.init_logger(__name__)
+
+
+@clouds.CLOUD_REGISTRY.register
+class IBM(clouds.Cloud):
+    """IBM Web Services."""
+
+    _REPR = 'IBM'
+    # IBM's VPC instance name length is limited to 63 chars.
+    # since Ray names the instances using the following
+    # format: ray-{cluster_name}-{node_type}-{8-char-uuid}
+    # it leaves 63-3-5-8=47 characters for the cluster name.
+    _MAX_CLUSTER_NAME_LEN_LIMIT = 47
+    _regions: List[clouds.Region] = []
+
+    @classmethod
+    def _cloud_unsupported_features(
+            cls) -> Dict[clouds.CloudImplementationFeatures, str]:
+        return dict()
+
+    @classmethod
+    def _max_cluster_name_length(cls) -> Optional[int]:
+        return cls._MAX_CLUSTER_NAME_LEN_LIMIT
+
+    @classmethod
+    def regions_with_offering(cls, instance_type: str,
+                              accelerators: Optional[Dict[str, int]],
+                              use_spot: bool, region: Optional[str],
+                              zone: Optional[str]) -> List[clouds.Region]:
+        del accelerators  # unused
+        assert use_spot is False, (
+            'current IBM implementation doesn\'t support spot instances')
+        regions = service_catalog.get_region_zones_for_instance_type(
+            instance_type, use_spot, 'ibm')
+
+        if region is not None:
+            regions = [r for r in regions if r.name == region]
+        if zone is not None:
+            for r in regions:
+                assert r.zones is not None, r
+                r.set_zones([z for z in r.zones if z.name == zone])
+            regions = [r for r in regions if r.zones]
+        return regions
+
+    @classmethod
+    def zones_provision_loop(
+        cls,
+        *,
+        region: str,
+        num_nodes: int,
+        instance_type: str,
+        accelerators: Optional[Dict[str, int]] = None,
+        use_spot: bool = False,
+    ) -> Iterator[Optional[List[Zone]]]:
+        """Loops over (region, zones) to retry for provisioning.
+
+        returning a single zone list with its region,
+        since ibm cloud currently doesn't
+        support retries for list of zones.
+
+        Args:
+            instance_type: The instance type to provision.
+            accelerators: The accelerators to provision.
+            use_spot: Whether to use spot instances.
+        """
+        # del accelerators  # unused
+        del num_nodes  # unused
+        assert use_spot is False, (
+            'current IBM implementation doesn\'t support spot instances')
+
+        regions = cls.regions_with_offering(instance_type,
+                                            accelerators,
+                                            use_spot,
+                                            region=region,
+                                            zone=None)
+        # IBM provisioner currently takes 1 zone per request.
+        for r in regions:
+            assert r.zones is not None, r
+            for zone in r.zones:
+                yield [zone]
+
+    @classmethod
+    def get_zone_shell_cmd(cls) -> Optional[str]:
+        return None
+
+    def instance_type_to_hourly_cost(self,
+                                     instance_type: str,
+                                     use_spot: bool,
+                                     region: Optional[str] = None,
+                                     zone: Optional[str] = None) -> float:
+        # Currently doesn't support spot instances, hence use_spot set to False.
+        del use_spot
+        return service_catalog.get_hourly_cost(instance_type,
+                                               use_spot=False,
+                                               region=region,
+                                               zone=zone,
+                                               clouds='ibm')
+
+    def accelerators_to_hourly_cost(self,
+                                    accelerators: Dict[str, int],
+                                    use_spot: bool,
+                                    region: Optional[str] = None,
+                                    zone: Optional[str] = None) -> float:
+        del accelerators, use_spot, region, zone  # unused
+        # Currently Isn't implemented in the same manner by aws and azure.
+        return 0
+
+    def get_egress_cost(self, num_gigabytes):
+        """Returns the egress cost. Currently true for us-south, i.e. Dallas.
+        based on https://cloud.ibm.com/objectstorage/create#pricing. """
+        cost = 0
+        price_thresholds = [{
+            'threshold': 150,
+            'price_per_gb': 0.05
+        }, {
+            'threshold': 50,
+            'price_per_gb': 0.07
+        }, {
+            'threshold': 0,
+            'price_per_gb': 0.09
+        }]
+        for price_threshold in price_thresholds:
+            # pylint: disable=line-too-long
+            cost += (num_gigabytes - price_threshold['threshold']
+                    ) * price_threshold['price_per_gb']
+            num_gigabytes -= (num_gigabytes - price_threshold['threshold'])
+        return cost
+
+    def is_same_cloud(self, other):
+        return isinstance(other, IBM)
+
+    def make_deploy_resources_variables(
+        self,
+        resources: 'resources_lib.Resources',
+        region: 'clouds.Region',
+        zones: Optional[List['clouds.Zone']],
+    ) -> Dict[str, Optional[str]]:
+        """Converts planned sky.Resources to cloud-specific resource variables.
+
+        These variables are used to fill the node type section (instance type,
+        any accelerators, etc.) in the cloud's deployment YAML template.
+
+        Cloud-agnostic sections (e.g., commands to run) need not be returned by
+        this function.
+
+        Returns:
+          A dictionary of cloud-specific node type variables.
+        """
+
+        def _get_profile_resources(instance_profile):
+            """returns a dict representing the
+             cpu, memory and gpu of specified instance profile"""
+
+            cpu_num, memory_num = self.get_vcpus_mem_from_instance_type(
+                instance_profile)
+            gpu_num = self.get_accelerators_from_instance_type(instance_profile)
+            gpu_num = list(gpu_num.values())[0] if gpu_num else 0
+            return {'CPU': cpu_num, 'memory': memory_num, 'GPU': gpu_num}
+
+        region_name = region.name
+        # zones is guaranteed to be initialized for
+        # clouds implementing 'zones_provision_loop()'
+        zone_names = [zone.name for zone in zones]  # type: ignore[union-attr]
+
+        r = resources
+        assert not r.use_spot, \
+            'IBM does not currently support spot instances in this framework'
+
+        acc_dict = self.get_accelerators_from_instance_type(r.instance_type)
+        if acc_dict is not None:
+            custom_resources = json.dumps(acc_dict, separators=(',', ':'))
+        else:
+            custom_resources = None
+
+        instance_resources = _get_profile_resources(r.instance_type)
+
+        worker_instance_type = get_cred_file_field('worker_instance_type',
+                                                   r.instance_type)
+        worker_instance_resources = _get_profile_resources(worker_instance_type)
+        # r.image_id: {clouds.Region:image_id} - property of Resources class
+        image_id = r.image_id[
+            region.name] if r.image_id else self.get_default_image(region_name)
+
+        return {
+            'instance_type': r.instance_type,
+            'instance_resources': instance_resources,
+            'worker_instance_type': worker_instance_type,
+            'worker_instance_resources': worker_instance_resources,
+            'custom_resources': custom_resources,
+            'use_spot': r.use_spot,
+            'region': region_name,
+            'zones': ','.join(zone_names),
+            'image_id': image_id,
+            'iam_api_key': get_cred_file_field('iam_api_key'),
+            'resource_group_id': get_cred_file_field('resource_group_id'),
+            'disk_capacity': get_cred_file_field('disk_capacity', 100)
+        }
+
+    @classmethod
+    def get_vcpus_mem_from_instance_type(
+        cls,
+        instance_type: str,
+    ) -> Tuple[Optional[float], Optional[float]]:
+        return service_catalog.get_vcpus_mem_from_instance_type(instance_type,
+                                                                clouds='ibm')
+
+    @classmethod
+    def get_accelerators_from_instance_type(
+        cls,
+        instance_type: str,
+    ) -> Optional[Dict[str, int]]:
+        """Returns {acc: acc_count} held by 'instance_type', if any."""
+        return service_catalog.get_accelerators_from_instance_type(
+            instance_type, clouds='ibm')
+
+    @classmethod
+    def get_default_instance_type(
+            cls,
+            cpus: Optional[str] = None,
+            memory: Optional[str] = None,
+            disk_tier: Optional[str] = None) -> Optional[str]:
+        return service_catalog.get_default_instance_type(cpus=cpus,
+                                                         memory=memory,
+                                                         disk_tier=disk_tier,
+                                                         clouds='ibm')
+
+    def get_feasible_launchable_resources(self,
+                                          resources: 'resources_lib.Resources'):
+        """Returns a list of feasible and launchable resources.
+
+        Feasible resources refer to an offering respecting the resource
+        requirements.  Currently, this function implements "filtering" the
+        cloud's offerings only w.r.t. accelerators constraints.
+
+        Launchable resources require a cloud and an instance type be assigned.
+        """
+        fuzzy_candidate_list: Optional[List[str]] = []
+        if resources.instance_type is not None:
+            assert resources.is_launchable(), resources
+            resources = resources.copy(accelerators=None)
+            return ([resources], fuzzy_candidate_list)
+
+        def _make(instance_list):
+            resource_list = []
+            for instance_type in instance_list:
+                r = resources.copy(
+                    cloud=IBM(),
+                    instance_type=instance_type,
+                    # Setting this to None as IBM doesn't separately bill /
+                    # attach the accelerators.  Billed as part of the VM type.
+                    accelerators=None,
+                    cpus=None,
+                    memory=None)
+                resource_list.append(r)
+            return resource_list
+
+        # Currently, handle a filter on accelerators only.
+        accelerators = resources.accelerators
+        if accelerators is None:
+            # Return a default instance type with the given number of vCPUs.
+            default_instance_type = IBM.get_default_instance_type(
+                cpus=resources.cpus,
+                memory=resources.memory,
+                disk_tier=resources.disk_tier)
+            if default_instance_type is None:
+                return ([], [])
+            else:
+                return (_make([default_instance_type]), [])
+
+        assert len(accelerators) == 1, resources
+        acc, acc_count = list(accelerators.items())[0]
+        (instance_list, fuzzy_candidate_list
+        ) = service_catalog.get_instance_type_for_accelerator(
+            acc,
+            acc_count,
+            cpus=resources.cpus,
+            memory=resources.memory,
+            region=resources.region,
+            zone=resources.zone,
+            clouds='ibm')
+        if instance_list is None:
+            return ([], fuzzy_candidate_list)
+        return (_make(instance_list), fuzzy_candidate_list)
+
+    @classmethod
+    def get_default_image(cls, region) -> str:
+        """
+        Returns default image id, currently stock ubuntu 22-04.
+        if user specified 'image_id' in ~/.ibm/credentials.yaml
+            matching this 'region', returns it instead.
+        """
+
+        def _get_image_objects():
+            # pylint: disable=E1136
+            images = []
+            res = client.list_images().get_result()
+            images.extend(res['images'])
+
+            while res.get('next'):
+                link_to_next = res['next']['href'].split('start=')[1].split(
+                    '&limit')[0]
+                res = client.list_images(start=link_to_next).get_result()
+                images.extend(res['images'])
+            return images
+
+        # if user specified an image id for the region, return it.
+        # IBM-TODO take into account user input when --image-id
+        # is implemented
+        user_image = get_cred_file_field('image_id')
+        if user_image and region in user_image:
+            logger.debug(f'using user image: {user_image[region]} '
+                         f'in region: {region}.')
+            return user_image[region]
+
+        client = ibm.client(region=region)
+        # returns default image: "ibm-ubuntu-22-04" with amd architecture
+        return next((img for img in _get_image_objects() if
+         img['name'].startswith('ibm-ubuntu-22-04') \
+            and img['operating_system']['architecture'].startswith(
+                'amd')))['id']
+
+    def get_image_size(self, image_id: str, region: Optional[str]) -> float:
+        assert region is not None, (image_id, region)
+        client = ibm.client(region=region)
+        try:
+            image_data = client.get_image(image_id).get_result()
+        # pylint: disable=line-too-long
+        except ibm.ibm_cloud_sdk_core.ApiException as e:  # type: ignore[union-attr]
+            logger.error(e.message)
+            with ux_utils.print_exception_no_traceback():
+                raise ValueError(f'Image {image_id!r} not found in '
+                                 f'IBM region "{region}"') from None
+        try:
+            # image_size['file']['size'] is not relevant, since
+            # the minimum size of a volume onto which this image
+            # may be provisioned is stored in minimum_provisioned_size
+            # pylint: disable=unsubscriptable-object
+            image_size = image_data['minimum_provisioned_size']
+        except KeyError:
+            logger.error('Image missing metadata:"minimum_provisioned_size". '
+                         'Image may be in status: Failed/Pending.')
+            with ux_utils.print_exception_no_traceback():
+                raise ValueError(
+                    f'IBM image {image_id!r} in '
+                    f'region "{region}", is missing'
+                    'metadata:"minimum_provisioned_size". '
+                    'Image may be in status: Failed/Pending') from None
+        return image_size
+
+    @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']
+        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 = _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)
+
+        # verifies ability of user to create a client,
+        # e.g. bad API KEY.
+        try:
+            ibm.client()
+            return True, None
+        # pylint: disable=W0703
+        except Exception as e:
+            return (False, f'{str(e)}' + help_str)
+
+    @classmethod
+    def check_disk_tier_enabled(cls, instance_type: str,
+                                disk_tier: str) -> None:
+        del instance_type, disk_tier  # unused
+
+    def get_credential_file_mounts(self) -> Dict[str, str]:
+        """Returns a {remote:local} credential path mapping
+         written to the cluster's file_mounts segment
+         of its yaml file (e.g., ibm-ray.yml.j2)
+        """
+        return {CREDENTIAL_FILE: CREDENTIAL_FILE}
+
+    def instance_type_exists(self, instance_type):
+        """Returns whether the instance type exists for this cloud."""
+        return service_catalog.instance_type_exists(instance_type, clouds='ibm')
+
+    def validate_region_zone(self, region: Optional[str], zone: Optional[str]):
+        """Validates the region and zone."""
+        return service_catalog.validate_region_zone(region, zone, clouds='ibm')
+
+    def accelerator_in_region_or_zone(self,
+                                      accelerator: str,
+                                      acc_count: int,
+                                      region: Optional[str] = None,
+                                      zone: Optional[str] = None) -> bool:
+        """Returns whether the accelerator is valid in the region or zone."""
+        return service_catalog.accelerator_in_region_or_zone(
+            accelerator, acc_count, region, zone, 'ibm')
+
+
+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()
+    if not base_config:
+        raise FileNotFoundError('Missing '
+                                f'credential file at {CREDENTIAL_FILE}')
+    return base_config.get(field, default_val)
diff --git a/sky/clouds/service_catalog/__init__.py b/sky/clouds/service_catalog/__init__.py
index ef003bb8a..81884ddd0 100644
--- a/sky/clouds/service_catalog/__init__.py
+++ b/sky/clouds/service_catalog/__init__.py
@@ -16,7 +16,7 @@ if typing.TYPE_CHECKING:
     from sky.clouds.service_catalog import common
 
 CloudFilter = Optional[Union[List[str], str]]
-_ALL_CLOUDS = ('aws', 'azure', 'gcp', 'lambda')
+_ALL_CLOUDS = ('aws', 'azure', 'gcp', 'ibm', 'lambda')
 
 
 def _map_clouds_catalog(clouds: CloudFilter, method_name: str, *args, **kwargs):
diff --git a/sky/clouds/service_catalog/ibm_catalog.py b/sky/clouds/service_catalog/ibm_catalog.py
new file mode 100644
index 000000000..cf5b47657
--- /dev/null
+++ b/sky/clouds/service_catalog/ibm_catalog.py
@@ -0,0 +1,122 @@
+"""
+IBM Offerings Catalog.
+
+This module loads the service catalog file and can be used to query
+instance types and pricing information for IBM.
+"""
+
+from sky import sky_logging
+from sky.clouds import cloud
+from sky.clouds.service_catalog import common
+from sky.adaptors import ibm
+from typing import Dict, List, Optional, Tuple
+
+logger = sky_logging.init_logger(__name__)
+
+_DEFAULT_INSTANCE_FAMILY = 'bx2'
+_DEFAULT_NUM_VCPUS = '8'
+_DEFAULT_MEMORY = 32
+
+_df = common.read_catalog('ibm/vms.csv')
+
+
+def instance_type_exists(instance_type: str) -> bool:
+    return common.instance_type_exists_impl(_df, instance_type)
+
+
+def validate_region_zone(region: Optional[str], zone: Optional[str]):
+    return common.validate_region_zone_impl('IBM', _df, region, zone)
+
+
+def accelerator_in_region_or_zone(acc_name: str,
+                                  acc_count: int,
+                                  region: Optional[str] = None,
+                                  zone: Optional[str] = None) -> bool:
+    return common.accelerator_in_region_or_zone_impl(_df, acc_name, acc_count,
+                                                     region, zone)
+
+
+def get_hourly_cost(instance_type: str,
+                    use_spot: bool = False,
+                    region: Optional[str] = None,
+                    zone: Optional[str] = None) -> float:
+    return common.get_hourly_cost_impl(_df, instance_type, use_spot, region,
+                                       zone)
+
+
+def get_vcpus_mem_from_instance_type(
+        instance_type: str) -> Tuple[Optional[float], Optional[float]]:
+    return common.get_vcpus_mem_from_instance_type_impl(_df, instance_type)
+
+
+def get_accelerators_from_instance_type(
+        instance_type: str) -> Optional[Dict[str, int]]:
+    return common.get_accelerators_from_instance_type_impl(_df, instance_type)
+
+
+def get_instance_type_for_accelerator(
+    acc_name: str,
+    acc_count: int,
+    cpus: Optional[str] = None,
+    memory: Optional[str] = None,
+    use_spot: bool = False,
+    region: Optional[str] = None,
+    zone: Optional[str] = None,
+) -> Tuple[Optional[List[str]], List[str]]:
+    """
+    Returns a list of instance types satisfying the required count of
+    accelerators with sorted prices and a list of candidates with fuzzy search.
+    """
+    return common.get_instance_type_for_accelerator_impl(df=_df,
+                                                         acc_name=acc_name,
+                                                         acc_count=acc_count,
+                                                         cpus=cpus,
+                                                         memory=memory,
+                                                         use_spot=use_spot,
+                                                         region=region,
+                                                         zone=zone)
+
+
+def get_region_zones_for_instance_type(instance_type: str,
+                                       use_spot: bool) -> List[cloud.Region]:
+    df = _df[_df['InstanceType'] == instance_type]
+    return common.get_region_zones(df, use_spot)
+
+
+def list_accelerators(
+        gpus_only: bool,
+        name_filter: Optional[str],
+        region_filter: Optional[str],
+        case_sensitive: bool = True
+) -> Dict[str, List[common.InstanceTypeInfo]]:
+    """Returns all instance types in IBM offering accelerators."""
+    return common.list_accelerators_impl('IBM', _df, gpus_only, name_filter,
+                                         region_filter, case_sensitive)
+
+
+def get_default_instance_type(cpus: Optional[str] = None,
+                              memory: Optional[str] = None,
+                              disk_tier: Optional[str] = None) -> Optional[str]:
+    del disk_tier  # unused
+    if cpus is None and memory is None:
+        cpus = f'{_DEFAULT_NUM_VCPUS}+'
+
+    if memory is None:
+        memory_gb_or_ratio = f'{_DEFAULT_MEMORY}+'
+    else:
+        memory_gb_or_ratio = memory
+    instance_type_prefix = f'{_DEFAULT_INSTANCE_FAMILY}-'
+    df = _df[_df['InstanceType'].str.startswith(instance_type_prefix)]
+    return common.get_instance_type_for_cpus_mem_impl(df, cpus,
+                                                      memory_gb_or_ratio)
+
+
+def is_image_tag_valid(tag: str, region: Optional[str]) -> bool:
+    """Returns whether the image tag is valid."""
+    vpc_client = ibm.client(region=region)
+    try:
+        vpc_client.get_image(tag)
+    except ibm.ibm_cloud_sdk_core.ApiException as e:  # type: ignore[union-attr]
+        logger.error(e.message)
+        return False
+    return True
diff --git a/sky/resources.py b/sky/resources.py
index 453f4a0b3..a7e8cfc94 100644
--- a/sky/resources.py
+++ b/sky/resources.py
@@ -648,10 +648,12 @@ class Resources:
                     'Cloud must be specified when image_id is provided.')
 
         if not self._cloud.is_same_cloud(
-                clouds.AWS()) and not self._cloud.is_same_cloud(clouds.GCP()):
+                clouds.IBM()) and not self._cloud.is_same_cloud(
+                    clouds.AWS()) and not self._cloud.is_same_cloud(
+                        clouds.GCP()):
             with ux_utils.print_exception_no_traceback():
                 raise ValueError(
-                    'image_id is only supported for AWS and GCP, please '
+                    'image_id is only supported for AWS, GCP and IBM, please '
                     'explicitly specify the cloud.')
 
         if self._region is not None:
@@ -897,7 +899,7 @@ class Resources:
             resources_fields['zone'] = config.pop('zone')
         if config.get('image_id') is not None:
             logger.warning('image_id in resources is experimental. It only '
-                           'supports AWS/GCP.')
+                           'supports AWS/GCP/IBM.')
             resources_fields['image_id'] = config.pop('image_id')
         if config.get('disk_tier') is not None:
             resources_fields['disk_tier'] = config.pop('disk_tier')
diff --git a/sky/setup_files/MANIFEST.in b/sky/setup_files/MANIFEST.in
index 9a3e6a2b8..dda84b153 100644
--- a/sky/setup_files/MANIFEST.in
+++ b/sky/setup_files/MANIFEST.in
@@ -5,6 +5,7 @@ include sky/skylet/providers/aws/cloudwatch/*
 include sky/skylet/providers/azure/*
 include sky/skylet/providers/gcp/*
 include sky/skylet/providers/lambda_cloud/*
+include sky/skylet/providers/ibm/*
 include sky/skylet/ray_patches/*.patch
 include sky/templates/*
 include sky/setup_files/*
diff --git a/sky/setup_files/setup.py b/sky/setup_files/setup.py
index 087aa1af3..0d3d0b31c 100644
--- a/sky/setup_files/setup.py
+++ b/sky/setup_files/setup.py
@@ -123,6 +123,7 @@ 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'],
     'docker': ['docker'],
     'lambda': [],
     'cloudflare': aws_dependencies
diff --git a/sky/skylet/events.py b/sky/skylet/events.py
index 20de332f1..c3a76222e 100644
--- a/sky/skylet/events.py
+++ b/sky/skylet/events.py
@@ -139,15 +139,24 @@ class AutostopEvent(SkyletEvent):
             self._replace_yaml_for_stopping(self._ray_yaml_path,
                                             autostop_config.down)
 
-            # We do "initial ray up + ray down --workers-only" only for
-            # multinode clusters for two reasons: (1) optimization: they are
-            # not needed for single-node; (2) for single-node SSO clusters: we
+            # Use environment variables to disable the ray usage collection (to
+            # avoid overheads and potential issues with the usage) as sdk does
+            # not take the argument for disabling the usage collection.
+            #
+            # Also clear any cloud-specific credentials set as env vars (e.g.,
+            # AWS's two env vars). Reason: for single-node AWS SSO clusters, we
             # have seen a weird bug where user image's /etc/profile.d may
             # contain the two AWS env vars, and so they take effect in the
-            # initial 'ray up', throwing a RuntimeError when some private VPC
-            # is not found (since it only exists in the assumed role, not in
-            # the custome principal set by the env vars). We use both this
-            # optimization and removing inheriting os.environ to fix this.
+            # bootstrap phase of each of these 3 'ray' commands, throwing a
+            # RuntimeError when some private VPC is not found (since the VPC
+            # only exists in the assumed role, not in the custome principal set
+            # by the env vars).  See #1880 for details.
+            env = dict(os.environ, RAY_USAGE_STATS_ENABLED='0')
+            env.pop('AWS_ACCESS_KEY_ID', None)
+            env.pop('AWS_SECRETE_ACCESS_KEY', None)
+
+            # We do "initial ray up + ray down --workers-only" only for
+            # multinode clusters as they are not needed for single-node.
             if is_cluster_multinode:
                 # `ray up` is required to reset the upscaling speed and min/max
                 # workers. Otherwise, `ray down --workers-only` will
@@ -157,30 +166,30 @@ class AutostopEvent(SkyletEvent):
                           write_ray_up_script_with_patched_launch_hash_fn(
                               self._ray_yaml_path,
                               ray_up_kwargs={'restart_only': True}))
-                subprocess.run(
-                    [sys.executable, script],
-                    check=True,
-                    # Use environment variables to disable the ray usage
-                    # collection (to avoid overheads and potential issues with
-                    # the usage) as sdk does not take the argument for
-                    # disabling the usage collection.
-                    #
-                    # Do not inherit from os.environ, which may have
-                    # cloud-specific credentials set as env vars (e.g., AWS's
-                    # two env vars). See #1880 and comments above for details.
-                    env=dict(RAY_USAGE_STATS_ENABLED='0'),
-                )
+                # Passing env inherited from os.environ is technically not
+                # needed, because we call `python <script>` rather than `ray
+                # <cmd>`. We just need the {RAY_USAGE_STATS_ENABLED: 0} part.
+                subprocess.run([sys.executable, script], check=True, env=env)
 
                 logger.info('Running ray down.')
                 # Stop the workers first to avoid orphan workers.
-                subprocess.run([
-                    'ray', 'down', '-y', '--workers-only', self._ray_yaml_path
-                ],
-                               check=True)
+                subprocess.run(
+                    [
+                        'ray', 'down', '-y', '--workers-only',
+                        self._ray_yaml_path
+                    ],
+                    check=True,
+                    # We pass env inherited from os.environ due to calling `ray
+                    # <cmd>`.
+                    env=env)
 
             logger.info('Running final ray down.')
-            subprocess.run(['ray', 'down', '-y', self._ray_yaml_path],
-                           check=True)
+            subprocess.run(
+                ['ray', 'down', '-y', self._ray_yaml_path],
+                check=True,
+                # We pass env inherited from os.environ due to calling `ray
+                # <cmd>`.
+                env=env)
         else:
             raise NotImplementedError
 
diff --git a/sky/skylet/providers/ibm/__init__.py b/sky/skylet/providers/ibm/__init__.py
new file mode 100644
index 000000000..21d789ce2
--- /dev/null
+++ b/sky/skylet/providers/ibm/__init__.py
@@ -0,0 +1,2 @@
+"""IBM node provider"""
+from sky.skylet.providers.ibm.node_provider import IBMVPCNodeProvider
diff --git a/sky/skylet/providers/ibm/node_provider.py b/sky/skylet/providers/ibm/node_provider.py
new file mode 100644
index 000000000..068594d9f
--- /dev/null
+++ b/sky/skylet/providers/ibm/node_provider.py
@@ -0,0 +1,955 @@
+#
+# (C) Copyright IBM Corp. 2021
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+""" module allocating nodes to Ray's cluster.
+ used by the Ray framework as an external connector to IBM provider. """
+
+import concurrent.futures as cf
+import inspect
+import json
+import re
+import socket
+import threading
+import time
+from pprint import pprint
+from pathlib import Path
+from uuid import uuid4
+
+from sky.adaptors import ibm
+from typing import Any, Dict, List, Optional
+
+from ray.autoscaler._private.cli_logger import cli_logger
+from ray.autoscaler._private.util import hash_runtime_conf, hash_launch_conf
+from ray.autoscaler.node_provider import NodeProvider
+from ray.autoscaler.tags import (
+    NODE_KIND_HEAD,
+    NODE_KIND_WORKER,
+    TAG_RAY_CLUSTER_NAME,
+    TAG_RAY_NODE_KIND,
+    TAG_RAY_NODE_NAME,
+    TAG_RAY_LAUNCH_CONFIG,
+    TAG_RAY_RUNTIME_CONFIG,
+    TAG_RAY_USER_NODE_TYPE,
+    TAG_RAY_NODE_STATUS,
+    TAG_RAY_FILE_MOUNTS_CONTENTS,
+)
+from sky.skylet.providers.ibm.vpc_provider import IBMVPCProvider
+from sky.skylet.providers.ibm.utils import get_logger, RAY_RECYCLABLE
+
+logger = get_logger("node_provider_")
+
+INSTANCE_NAME_UUID_LEN = 8
+INSTANCE_NAME_MAX_LEN = 64
+PENDING_TIMEOUT = 3600  # period before a node that's yet to run is deleted.
+VOLUME_TIER_NAME_DEFAULT = "general-purpose"
+# identifies resources created by this package.
+# these resources are deleted alongside the node.
+VPC_TAGS = ".sky-vpc-tags"
+
+
+def log_in_out(func):
+    """Tracing decorator for debugging purposes"""
+
+    def decorated_func(*args, **kwargs):
+        name = func.__name__
+        logger.debug(
+            f"\n\nEnter {name} from {inspect.stack()[0][3]} "
+            f"{inspect.stack()[1][3]} {inspect.stack()[2][3]} with args: "
+            f"entered with args:\n{pprint(args)} and kwargs {pprint(kwargs)}"
+        )
+        try:
+            result = func(*args, **kwargs)
+            logger.debug(
+                f"Leave {name} from {inspect.stack()[1][3]} with result "
+                f"Func Result:{pprint(result)}\n\n"
+            )
+        except Exception:
+            cli_logger.error(f"Error in {name}")
+            raise
+        return result
+
+    return decorated_func
+
+
+class IBMVPCNodeProvider(NodeProvider):
+    """Node Provider for IBM VPC
+
+    Currently, instance tagging is implemented using internal cache
+
+    To communicate with head node from outside cluster private network,
+    `use_hybrid_ips` set to True. Then, floating (external) ip allocated to
+    cluster head node, while worker nodes are provisioned with private ips only.
+    """
+
+    # IBM-TODO alter tagging mechanism to support cloud tags instead of
+    #  maintaining VM tags locally (file).
+    def _load_tags(self):
+        """
+        if local tags cache (file) exists (cluster is restarting),
+            cache is loaded, deleted nodes are filtered away and result
+            is dumped to local cache.
+        otherwise, remote head node will initialize the in memory
+          and local storage tags cache with the head's cluster tags.
+        """
+
+        # local tags cache exists from former runs
+        if self.tags_file.is_file():
+            all_tags = json.loads(self.tags_file.read_text())
+            tags = all_tags.get(self.cluster_name, {})
+
+            # filters instances that were deleted since
+            # the last time the head node was up
+            for instance_id, instance_tags in tags.items():
+                try:
+                    self.ibm_vpc_client.get_instance(instance_id)
+                    self.nodes_tags[instance_id] = instance_tags
+                except ibm.ibm_cloud_sdk_core.ApiException as e:
+                    cli_logger.warning(instance_id)
+                    if e.message == "Instance not found":
+                        logger.error(
+                            f"cached instance {instance_id} not found, \
+                                will be removed from cache"
+                        )
+            # dump in-memory cache to local cache (file).
+            self.set_node_tags(None, None)
+
+        else:
+            name = socket.gethostname()  # returns the instance's (VSI) name
+            logger.debug(f"Check if {name} is HEAD")
+            # check if current VM is the remote head node
+            if self._get_node_type(name) == NODE_KIND_HEAD:
+                logger.debug(f"{name} is HEAD")
+                # pylint: disable=line-too-long
+                node = self.ibm_vpc_client.list_instances(name=name).get_result()[
+                    "instances"
+                ]
+                if node:
+                    logger.debug(f"{name} is node in vpc")
+
+                    # remote head node will calculate 2 hash values below
+                    ray_bootstrap_config = Path.home() / "ray_bootstrap_config.yaml"
+                    config = json.loads(ray_bootstrap_config.read_text())
+                    (runtime_hash, mounts_contents_hash) = hash_runtime_conf(
+                        config["file_mounts"], None, config
+                    )
+
+                    # launch_hash is used to verify the head node belongs
+                    # to the cluster. e.g. checked when running --restart-only
+                    head_node_config = {}
+                    head_node_type = config.get("head_node_type")
+                    if head_node_type:
+                        head_config = config["available_node_types"][head_node_type]
+                        head_node_config.update(head_config["node_config"])
+                    launch_hash = hash_launch_conf(head_node_config, config["auth"])
+
+                    head_tags = {
+                        TAG_RAY_NODE_KIND: NODE_KIND_HEAD,
+                        TAG_RAY_NODE_NAME: name,
+                        TAG_RAY_NODE_STATUS: "up-to-date",
+                        TAG_RAY_CLUSTER_NAME: self.cluster_name,
+                        TAG_RAY_USER_NODE_TYPE: config["head_node_type"],
+                        TAG_RAY_RUNTIME_CONFIG: runtime_hash,
+                        TAG_RAY_LAUNCH_CONFIG: launch_hash,
+                        TAG_RAY_FILE_MOUNTS_CONTENTS: mounts_contents_hash,
+                    }
+
+                    logger.debug(f"Setting HEAD node tags {head_tags}")
+                    self.set_node_tags(node[0]["id"], head_tags)
+
+    def __init__(self, provider_config, cluster_name):
+        """
+        Args:
+            provider_config (dict): containing the provider segment of
+                the cluster's config file.initialized as an instance
+                variable of the parent class, hence can accessed via self.
+            cluster_name(str): value of cluster_name within the
+                 cluster's config file.
+
+        """
+        NodeProvider.__init__(self, provider_config, cluster_name)
+
+        self.lock = threading.RLock()
+        self.tags_file = Path.home() / VPC_TAGS
+        self.iam_api_key = provider_config["iam_api_key"]
+        self.resource_group_id = provider_config["resource_group_id"]
+        self.iam_endpoint = provider_config.get("iam_endpoint")
+        self.region = provider_config["region"]
+        self.zone = self.provider_config["availability_zone"]
+        self.ibm_vpc_client = ibm.client(region=self.region)
+
+        # vpc_tags may contain the following fields:
+        # vpc_id, subnet_id, security_group_id
+        self.vpc_tags = None
+        self.vpc_provider = IBMVPCProvider(
+            self.resource_group_id, self.region, self.cluster_name
+        )
+        # cache of the cluster's nodes in the format: {node_id:node_tags}
+        self.nodes_tags = {}
+        # Cache of starting/running/pending(below PENDING_TIMEOUT)nodes.
+        # in the format{node_id:node_data}.
+        self.cached_nodes = {}
+        # cache of the nodes created, but not yet tagged and running.
+        # in the format {node_id:time_of_creation}.
+        self.pending_nodes = {}
+        # ids of nodes scheduled for deletion.
+        self.deleted_nodes = []
+
+        self._load_tags()
+
+        # if cache_stopped_nodes == true, nodes will be stopped
+        #   instead of deleted to better accommodate future rise in demand
+        self.cache_stopped_nodes = provider_config.get("cache_stopped_nodes", True)
+
+    def _get_node_type(self, name):
+        """returns the type of node with the specified name.
+
+        Args:
+            name (str): node's name
+        Returns:
+            node type (str): NODE_KIND_HEAD or NODE_KIND_WORKER.
+            None: return none if node name doesn't belong to cluster's
+                name scheme.
+        """
+        if f"{self.cluster_name}-{NODE_KIND_WORKER}" in name:
+            return NODE_KIND_WORKER
+        elif f"{self.cluster_name}-{NODE_KIND_HEAD}" in name:
+            return NODE_KIND_HEAD
+
+    def _get_nodes_by_tags(self, filters):
+        """
+        returns list of nodes' instance data who's tags are matching the specified filters.
+        updates self.nodes_tags if no filters were specified or
+          the only filter is the node's type.
+
+        This function is used by non_terminate_nodes to update self.cached_nodes
+          and self.nodes_tags thus abstains from using the cache. moreover,
+          using the cache from this function will create a loop on cache miss.
+        Args:
+            filters(dict):  specified conditions to filter nodes by.
+        Returns:
+            list of node data for each node that matches filters.
+        """
+
+        nodes = []
+        found_new_node = False
+        # filters specified are a subset of {cluster_name, node_type}
+        if {TAG_RAY_NODE_KIND, TAG_RAY_CLUSTER_NAME}.issuperset(set(filters.keys())):
+            if self.vpc_tags:
+                # if possible, search within the vpc to narrow the search scope
+                result = self.ibm_vpc_client.list_instances(
+                    vpc_id=self.vpc_tags["vpc_id"]
+                ).get_result()
+            else:
+                result = self.ibm_vpc_client.list_instances().get_result()
+            instances = result["instances"]
+            # using pagination to acquire instances from all result pages:
+            # https://cloud.ibm.com/apidocs/vpc/latest?code=python#api-pagination
+            while result.get("next"):
+                start = result["next"]["href"].split("start=")[1]
+                result = self.ibm_vpc_client.list_instances(start=start).get_result()
+                instances.extend(result["instances"])
+
+            for instance in instances:
+                kind = self._get_node_type(instance["name"])
+                if kind and instance["id"] not in self.deleted_nodes:
+                    if not filters or kind == filters[TAG_RAY_NODE_KIND]:
+                        nodes.append(instance)
+                        with self.lock:
+                            node_tags = self.nodes_tags.setdefault(instance["id"], {})
+                            if not node_tags:
+                                found_new_node = True
+                            node_tags.update(
+                                {
+                                    TAG_RAY_CLUSTER_NAME: self.cluster_name,
+                                    TAG_RAY_NODE_KIND: kind,
+                                }
+                            )
+            # update local tags cache
+            if found_new_node:
+                self.set_node_tags(None, None)
+
+        # if more complex set of filters is specified, there's no use to search
+        # the VPC for them, as this implementation stores tags locally instead
+        # of tagging the instances. instead we'll search the cache nodes_tags
+        else:  # match filters specified
+            with self.lock:
+
+                for node_id, node_tags in self.nodes_tags.items():
+                    # skipping nodes that don't match all specified filters,
+                    # i.e. not all items in filters match those in self.nodes_tags
+                    if not all(item in node_tags.items() for item in filters.items()):
+                        logger.debug(
+                            f"specified filters: {filters} don't match node "
+                            f"tags: {node_tags}"
+                        )
+                        continue
+
+                    # append to result list nodes that do match the filters:
+                    try:
+                        # not using cache. see function docs for details.
+                        nodes.append(self.ibm_vpc_client.get_instance(node_id).result)
+                    except ibm.ibm_cloud_sdk_core.ApiException as e:
+                        cli_logger.warning(node_id)
+                        if "Instance not found" in e.message:
+                            logger.error(f"failed to find vsi {node_id}, skipping")
+                            continue
+                        logger.error(f"failed to find instance {node_id}, raising")
+                        raise e
+
+        return nodes
+
+    def non_terminated_nodes(self, tag_filters) -> List[str]:
+        """
+        returns list of ids of non terminated nodes,
+        matching the specified tags. updates the nodes cache.
+        NOTE: this function is called periodically by ray,
+            a fact utilized to refresh caches (self.cached_nodes and self.node_tags).
+        Args:
+            tag_filters(dict): fields by which nodes will be filtered.
+        """
+
+        # collecting valid nodes that are either starting, running or pending.
+        res_nodes = []
+
+        found_nodes = self._get_nodes_by_tags(tag_filters)
+
+        for node in found_nodes:
+
+            # check if node scheduled for delete
+            with self.lock:
+                if node["id"] in self.deleted_nodes:
+                    logger.info(f"{node['id']} scheduled for delete")
+                    continue
+
+            # delete failed nodes and skip nodes in other invalid states
+            valid_statuses = ["pending", "starting", "running"]
+            if node["status"] not in valid_statuses:
+                with self.lock:
+                    if node["status"] == "failed":
+                        # since non_terminated_nodes is called periodically,
+                        # prevent access until node is deleted
+                        self._delete_node(node["id"], node)
+                        logger.error(
+                            """Encountered a failed node.
+                            Region might be crowded,
+                            Consider retrying later."""
+                        )
+                    else:
+                        logger.debug(
+                            f"{node['id']} status {node['status']}"
+                            f" not in {valid_statuses}, skipping"
+                        )
+                        continue
+
+            # remove instance hanging in pending state
+            with self.lock:
+                if node["id"] in self.pending_nodes:
+                    if node["status"] != "running":
+                        pending_time = time.time() - self.pending_nodes[node["id"]]
+                        logger.debug(f"{node['id']} is pending for {pending_time}")
+                        if pending_time > PENDING_TIMEOUT:
+                            logger.error(
+                                f"pending timeout {PENDING_TIMEOUT} reached, "
+                                f"deleting instance {node['id']}"
+                            )
+                            self._delete_node(node["id"], node)
+                            # avoid adding the node to cached_nodes
+                            continue
+                    else:
+                        self.pending_nodes.pop(node["id"], None)
+
+            # verify a floating ip is attached to the node
+            nic_id = node["network_interfaces"][0]["id"]
+            # pylint: disable=line-too-long
+            res = self.ibm_vpc_client.list_instance_network_interface_floating_ips(
+                node["id"], nic_id
+            ).get_result()
+            floating_ips = res["floating_ips"]
+            if len(floating_ips) == 0:
+                # not adding a node that's yet/failed to
+                # to get a floating ip provisioned
+                continue
+            else:
+                node["floating_ips"] = floating_ips
+
+            res_nodes.append(node)
+
+        # update cached_nodes
+        for node in res_nodes:
+            self.cached_nodes[node["id"]] = node
+
+        return [node["id"] for node in res_nodes]
+
+    def is_running(self, node_id) -> bool:
+        """returns whether a node is in status running"""
+        with self.lock:
+            node = self._get_cached_node(node_id)
+            logger.debug(
+                f"""node: {node_id} is_running?
+                {node["status"] == "running"}"""
+            )
+            return node["status"] == "running"
+
+    def is_terminated(self, node_id) -> bool:
+        """returns True if a node is either not recorded or
+        not in any valid status."""
+        with self.lock:
+            try:
+                node = self._get_cached_node(node_id)
+                logger.debug(
+                    f"""node: {node_id} is_terminated?
+                    {node["status"] not in
+                    ["running", "starting", "pending"]}"""
+                )
+                return node["status"] not in ["running", "starting", "pending"]
+            # pylint: disable=W0703
+            except Exception:
+                return True
+
+    def node_tags(self, node_id) -> Dict[str, str]:
+        """returns tags of specified node id"""
+
+        with self.lock:
+            return self.nodes_tags.get(node_id, {})
+
+    def external_ip(self, node_id) -> Optional[str]:
+        """returns node's public ip
+
+        Args:
+            node_id (str): id of a node within ibm VPC
+
+        Returns:
+            str: public ip of specified node
+        """
+
+        with self.lock:
+            node = self._get_cached_node(node_id)
+            fip = node.get("floating_ips")
+            if fip:
+                return fip[0]["address"]
+        return None
+
+    def internal_ip(self, node_id) -> str:
+        """returns the worker's node private ip address"""
+        node = self._get_cached_node(node_id)
+
+        # if a bug ocurred, or node data was fetched before primary_ip
+        # was assigned, refetch node data from cloud.
+        try:
+            primary_ip = node["network_interfaces"][0].get("primary_ip")["address"]
+            if primary_ip is None:
+                node = self._get_node(node_id)
+        # pylint: disable=W0703
+        except Exception:
+            node = self._get_node(node_id)
+
+        return node["network_interfaces"][0].get("primary_ip")["address"]
+
+    def set_node_tags(self, node_id, tags) -> None:
+        """
+        updates local (file) tags cache.
+        also updates in memory cache (self.nodes_tags) if node_id and tags
+        are specified.
+
+        Args:
+            node_id(str): node unique id.
+            tags(dict): specified conditions by which nodes will be filtered.
+        """
+
+        with self.lock:
+            # update in-memory cache
+            if node_id and tags:
+                node_tags = self.nodes_tags.setdefault(node_id, {})
+                node_tags.update(tags)
+
+            # dump in-memory cache to file.
+            all_tags = {}
+            if self.tags_file.is_file():
+                all_tags = json.loads(self.tags_file.read_text())
+            all_tags[self.cluster_name] = self.nodes_tags
+            self.tags_file.write_text(json.dumps(all_tags))
+
+    def _get_instance_data(self, name):
+        """Returns instance (node) information matching the specified name"""
+
+        instances_data = self.ibm_vpc_client.list_instances(name=name).get_result()
+        if len(instances_data["instances"]) > 0:
+            return instances_data["instances"][0]
+        return None
+
+    def _create_instance(self, name, base_config):
+        """
+        Creates a new VM instance with the specified name,
+        based on the provided base_config configuration dictionary
+        Args:
+            name(str): name of the instance.
+            base_config(dict): specific node relevant data.
+            node type segment of the cluster's config file,
+             e.g. ray_head_default.
+        """
+
+        logger.info(f"Creating new VM instance {name}")
+
+        security_group_identity_model = {"id": self.vpc_tags["security_group_id"]}
+        subnet_identity_model = {"id": self.vpc_tags["subnet_id"]}
+        primary_network_interface = {
+            "name": "eth0",
+            "subnet": subnet_identity_model,
+            "security_groups": [security_group_identity_model],
+        }
+
+        boot_volume_profile = {
+            "capacity": base_config.get("boot_volume_capacity", 100),
+            "name": f"boot-volume-{uuid4().hex[:4]}",
+            "profile": {
+                "name": base_config.get("volume_tier_name", VOLUME_TIER_NAME_DEFAULT)
+            },
+        }
+
+        boot_volume_attachment = {
+            "delete_volume_on_instance_delete": True,
+            "volume": boot_volume_profile,
+        }
+
+        key_identity_model = {"id": base_config["key_id"]}
+        profile_name = base_config.get("instance_profile_name")
+        if not profile_name:
+            raise Exception(
+                """Failed to detect 'instance_profile_name'
+                key in cluster config file"""
+            )
+
+        instance_prototype = {}
+        instance_prototype["name"] = name
+        instance_prototype["keys"] = [key_identity_model]
+        instance_prototype["profile"] = {"name": profile_name}
+        instance_prototype["resource_group"] = {"id": self.resource_group_id}
+        instance_prototype["vpc"] = {"id": self.vpc_tags["vpc_id"]}
+        instance_prototype["image"] = {"id": base_config["image_id"]}
+
+        instance_prototype["zone"] = {"name": self.zone}
+        instance_prototype["boot_volume_attachment"] = boot_volume_attachment
+        # pylint: disable=line-too-long
+        instance_prototype["primary_network_interface"] = primary_network_interface
+
+        try:
+            with self.lock:
+                resp = self.ibm_vpc_client.create_instance(instance_prototype)
+        except ibm.ibm_cloud_sdk_core.ApiException as e:
+            if e.code == 400 and "already exists" in e.message:
+                return self._get_instance_data(name)
+            elif e.code == 400 and "over quota" in e.message:
+                cli_logger.error(f"Create VM instance {name} failed due to quota limit")
+            else:
+                cli_logger.error(
+                    f"""Create VM instance for {name}
+                    failed with status code {str(e.code)}
+                    .\nFailed instance prototype:\n"""
+                )
+                pprint(instance_prototype)
+            raise e
+
+        logger.info(f"VM instance {name} created successfully")
+        return resp.result
+
+    def _create_floating_ip(self, base_config):
+        """returns unbound floating IP address.
+        Creates a new ip if none were found in the config file.
+        Args:
+            base_config(dict): specific node relevant data.
+            node type segment of the cluster's config file,
+            e.g. ray_head_default.
+        """
+
+        if base_config.get("head_ip"):
+            # pylint: disable=line-too-long
+            for ip in self.ibm_vpc_client.list_floating_ips().get_result()[
+                "floating_ips"
+            ]:
+                if ip["address"] == base_config["head_ip"]:
+                    return ip
+
+        floating_ip_name = f"{RAY_RECYCLABLE}-{uuid4().hex[:4]}"
+        # create a new floating ip
+        logger.debug(f"Creating floating IP {floating_ip_name}")
+        floating_ip_prototype = {}
+        floating_ip_prototype["name"] = floating_ip_name
+        floating_ip_prototype["zone"] = {"name": self.zone}
+        floating_ip_prototype["resource_group"] = {"id": self.resource_group_id}
+        response = self.ibm_vpc_client.create_floating_ip(floating_ip_prototype)
+        floating_ip_data = response.result
+
+        return floating_ip_data
+
+    def _attach_floating_ip(self, instance, fip_data):
+        """
+        attach a floating ip to the network interface of an instance
+        Args:
+            instance(dict): extensive data of a node.
+            fip_data(dict): floating ip data.
+        """
+
+        fip = fip_data["address"]
+        fip_id = fip_data["id"]
+
+        logger.debug(
+            f"""Attaching floating IP {fip} to
+            VM instance {instance["id"]}"""
+        )
+
+        # check if floating ip is not attached yet
+        inst_p_nic = instance["primary_network_interface"]
+
+        if inst_p_nic["primary_ip"] and inst_p_nic["id"] == fip_id:
+            # floating ip already attached. do nothing
+            logger.debug(f"Floating IP {fip} already attached to eth0")
+        else:
+            # attach floating ip
+            self.ibm_vpc_client.add_instance_network_interface_floating_ip(
+                instance["id"], instance["network_interfaces"][0]["id"], fip_id
+            )
+
+    def _stopped_nodes(self, tags):
+        """
+        returns either stopped head or stopped worker nodes,
+            belonging to this cluster
+
+        requires: TAG_RAY_NODE_KIND key in tags.
+        Args:
+            tags(dict): set of conditions nodes will be filtered by.
+        Returns:
+            list of stopped nodes, where each item contains node data.
+        """
+
+        # pylint: disable=C0206 W0622
+        # the following subset 'filter' is a common denominator for all
+        # nodes belonging to the cluster.
+        filter = {
+            TAG_RAY_CLUSTER_NAME: self.cluster_name,
+            TAG_RAY_NODE_KIND: tags[TAG_RAY_NODE_KIND],
+        }
+        nodes = []
+        # filter stopped nodes
+        with self.lock:
+            for node_id in self.nodes_tags:
+                try:
+                    node_tags = self.nodes_tags[node_id]
+                    if all(item in node_tags.items() for item in filter.items()):
+                        node = self.ibm_vpc_client.get_instance(node_id).result
+                        state = node["status"]
+                        if state in ["stopped", "stopping"]:
+                            nodes.append(node)
+                except Exception as e:
+                    cli_logger.warning(
+                        f"node: {node_id} from cache: 'nodes_tags' not found in VPC"
+                    )
+                    if (
+                        isinstance(e, ibm.ibm_cloud_sdk_core.ApiException)
+                        and "Instance not found" in e.message
+                    ):
+                        continue
+                    raise e
+            return nodes
+
+    def _create_node(self, base_config, tags):
+        """
+        returns dict {instance_id:instance_data} of newly created node.
+        updates tags cache.
+        creates a node in the following format:
+            ray-{cluster_name}-{node_type}-{uuid}
+
+        Args:
+            base_config(dict): specific node relevant data.
+                node type segment of the cluster's config file,
+                e.g. ray_head_default.
+            tags(dict): set of conditions nodes will be filtered by.
+        """
+
+        name_tag = tags[TAG_RAY_NODE_NAME]
+        if len(name_tag) > INSTANCE_NAME_MAX_LEN - INSTANCE_NAME_UUID_LEN - 1:
+            logger.error(
+                f"""node name: {name_tag} is longer than
+                {INSTANCE_NAME_MAX_LEN - INSTANCE_NAME_UUID_LEN - 1}
+                characters"""
+            )
+            raise Exception("Invalid node name: length check failed")
+
+        # IBM VPC VSI pattern requirement
+        pattern = re.compile("^([a-z]|[a-z][-a-z0-9]*[a-z0-9])$")
+        res = pattern.match(name_tag)
+
+        if not res:
+            logger.error(
+                f"""node name {name_tag} doesn't match the naming pattern
+                 `[a-z]|[a-z][-a-z0-9]*[a-z0-9]` for IBM VPC instance """
+            )
+            raise Exception(
+                "Invalid node name: pattern check for IBM VPC instance failed"
+            )
+
+        # append instance name with uuid
+        name = "{name_tag}-{uuid}".format(
+            name_tag=name_tag, uuid=uuid4().hex[:INSTANCE_NAME_UUID_LEN]
+        )
+
+        # create instance in vpc
+        instance = self._create_instance(name, base_config)
+
+        # record creation time. used to discover hanging nodes.
+        with self.lock:
+            self.pending_nodes[instance["id"]] = time.time()
+
+        tags[TAG_RAY_CLUSTER_NAME] = self.cluster_name
+        tags[TAG_RAY_NODE_NAME] = name
+        self.set_node_tags(instance["id"], tags)
+
+        # create and attach a floating ip to the node
+        fip_data = self._create_floating_ip(base_config)
+        self._attach_floating_ip(instance, fip_data)
+
+        return {instance["id"]: instance}
+
+    def create_node(self, base_config, tags, count) -> Dict[str, Any]:
+        """
+        returns dict of {instance_id:instance_data} of nodes.
+        creates 'count' number of nodes.
+        if enabled in base_config,
+        tries to re-run stopped nodes before creation of new instances.
+
+        Args:
+            base_config(dict): specific node relevant data.
+                node type segment of the cluster's config file,
+                e.g. ray_head_default.
+                a template shared by all nodes
+                when creating multiple nodes (count>1).
+            tags(dict): set of conditions nodes will be filtered by.
+            count(int): number of nodes to create.
+
+        """
+
+        # Create a VPC when creating a head node
+        # or fetch (for any node type) when restarting Ray
+        if not self.vpc_tags:
+            self.vpc_tags = self.vpc_provider.create_or_fetch_vpc(
+                self.region, self.zone
+            )
+
+        stopped_nodes_dict = {}
+        futures = []
+
+        # Try to reuse previously stopped nodes with compatible configs
+        if self.cache_stopped_nodes:
+            stopped_nodes = self._stopped_nodes(tags)
+            stopped_nodes_ids = [n["id"] for n in stopped_nodes]
+            stopped_nodes_dict = {n["id"]: n for n in stopped_nodes}
+
+            if stopped_nodes:
+                cli_logger.print(
+                    f"Reusing nodes {stopped_nodes_ids}. "
+                    "To disable reuse, set `cache_stopped_nodes: False` "
+                    "under `provider` in the cluster configuration."
+                )
+
+            for node in stopped_nodes:
+                logger.info(f"Starting instance {node['id']}")
+                self.ibm_vpc_client.create_instance_action(node["id"], "start")
+            # added delay due to a race condition.
+            # discussed at the final comment of this function.
+            time.sleep(1)
+
+            # update tags of stopped nodes
+            for node_id in stopped_nodes_ids:
+                self.set_node_tags(node_id, tags)
+
+            count -= len(stopped_nodes_ids)
+
+        created_nodes_dict = {}
+
+        # create multiple instances concurrently
+        if count:
+            with cf.ThreadPoolExecutor(count) as ex:
+                for _ in range(count):
+                    futures.append(ex.submit(self._create_node, base_config, tags))
+
+            for future in cf.as_completed(futures):
+                created_node = future.result()
+                created_nodes_dict.update(created_node)
+
+        all_created_nodes = stopped_nodes_dict
+        all_created_nodes.update(created_nodes_dict)
+
+        # this sleep is required due to race condition with non_terminated_nodes
+        # called in separate thread by autoscaler. not a lost as anyway the vsi
+        # operating system takes time to start. can be removed after
+        # https://github.com/ray-project/ray/issues/28150 resolved
+        time.sleep(5)
+
+        return all_created_nodes
+
+    def _delete_node(self, node_id, node_data=None):
+        """deletes a node and removes it from the caches.
+        if it's a (not failed) head node, deletes the vpc.
+
+        if node_data is specified uses it instead of fetching it
+        by using the node_id.
+
+        Args:
+            node_id (str): id of the node.
+            node_data (dict): node data received by the api: get_instance()
+
+        Raises:
+            ApiException: raises VPC api exception if deletion failed
+              for any reason other then "resource not found"
+        """
+
+        logger.info("Deleting VM instance {}".format(node_id))
+        # avoid fetching data if specified. Skipping this optimization
+        # could cause a loop if node is 'failed' since _get_cached_node()
+        # would eventually invoke non_terminated_nodes()
+        # which in turn will invoke this function.
+        node = node_data if node_data else self._get_cached_node(node_id)
+
+        # get vpc_id to delete if deleting head node which isn't a failed node
+        # pylint: disable=line-too-long
+        vpc_id_to_delete = None
+        if (
+            self._get_node_type(node["name"]) == NODE_KIND_HEAD
+            and node["status"] != "failed"
+        ):
+            if self.vpc_tags:
+                # if it was initialized it's sure to contain `vpc_id`
+                vpc_id_to_delete = self.vpc_tags["vpc_id"]
+            else:
+                vpc_id_to_delete = self.ibm_vpc_client.get_instance(
+                    node_id
+                ).get_result()["vpc"]["id"]
+
+            # if the request is executed by the remote head like in `sky autostop --down`
+            # , in contrast to `sky down`, use cloud functions to delete the VPC
+            # with its associated resources.
+            hostname = socket.gethostname()  # returns the instance's (VSI) name
+            if self._get_node_type(hostname) == NODE_KIND_HEAD:
+                self.vpc_provider.remote_cluster_removal(vpc_id_to_delete, self.region)
+
+        try:
+            self.ibm_vpc_client.delete_instance(node_id)
+
+            with self.lock:
+                # update node deletion in all caches
+                self.nodes_tags.pop(node_id, None)
+                self.pending_nodes.pop(node_id, None)
+                self.deleted_nodes.append(node_id)
+                self.cached_nodes.pop(node_id, None)
+
+                # calling set_node_tags with None
+                # will dump self.nodes_tags cache to file
+                self.set_node_tags(None, None)
+
+            # delete all ips created by this module
+            floating_ips = node.get("floating_ips", [])
+            for ip in floating_ips:
+                if ip["name"].startswith(RAY_RECYCLABLE):
+                    self.ibm_vpc_client.delete_floating_ip(ip["id"])
+
+        # catch exceptions from instance/ip deletion
+        except ibm.ibm_cloud_sdk_core.ApiException as e:
+            # allow 'resource not found' exceptions to pass through
+            # suggesting the resource was already removed
+            if e.code == 404:
+                pass
+            else:
+                raise e
+
+        # Terminate VPC if node deleted is a head node
+        if vpc_id_to_delete:
+            if self.poll_instance_deleted(node_id):
+                self.vpc_provider.delete_vpc(vpc_id_to_delete, self.region)
+
+    def terminate_nodes(self, node_ids):
+
+        if not node_ids:
+            return
+
+        futures = []
+        with cf.ThreadPoolExecutor(len(node_ids)) as ex:
+            for node_id in node_ids:
+                logger.debug(f"NodeProvider Terminating node {node_id}")
+                futures.append(ex.submit(self.terminate_node, node_id))
+
+        for future in cf.as_completed(futures):
+            future.result()
+
+    @log_in_out
+    def terminate_node(self, node_id):
+        """Deletes the VM instance and the associated volume.
+        if cache_stopped_nodes is set to true in the cluster config file,
+        nodes are stopped instead."""
+
+        try:
+            if self.cache_stopped_nodes:
+                cli_logger.print(
+                    f"Stopping instance {node_id}. To terminate instead, "
+                    "set `cache_stopped_nodes: False` "
+                    "under `provider` in the cluster configuration"
+                )
+
+                self.ibm_vpc_client.create_instance_action(node_id, "stop")
+            else:
+                cli_logger.print(f"Terminating instance {node_id}")
+                self._delete_node(node_id)
+
+        except ibm.ibm_cloud_sdk_core.ApiException as e:
+            if e.code == 404:
+                pass
+            else:
+                raise e
+
+    def _get_node(self, node_id):
+        """Refresh and get info for this node, updating the cache."""
+        err_msg = f"failed to get instance with id {node_id}"
+        # updates self.cached_nodes and nodes_tags
+        self.non_terminated_nodes({})
+
+        if node_id in self.cached_nodes:
+            return self.cached_nodes[node_id]
+        else:
+            logger.error(err_msg)
+            raise Exception(err_msg)
+
+    def _get_cached_node(self, node_id):
+        """Return node info from cache if possible, otherwise fetches it."""
+        if node_id in self.cached_nodes:
+            return self.cached_nodes[node_id]
+
+        return self._get_node(node_id)
+
+    def poll_instance_deleted(self, instance_id):
+        tries = 20  # waits up to 200sec with 10 sec interval
+        sleep_interval = 10
+        while tries:
+            try:
+                self.ibm_vpc_client.get_instance(instance_id).get_result()
+            except ibm.ibm_cloud_sdk_core.ApiException:
+                logger.debug(f"Deleted VM instance with id: {instance_id}")
+                return True
+
+            tries -= 1
+            time.sleep(sleep_interval)
+        logger.Error("\nFailed to delete instance within expected time frame\n")
+        return False
+
+    @staticmethod
+    def bootstrap_config(cluster_config) -> Dict[str, Any]:
+        return cluster_config
diff --git a/sky/skylet/providers/ibm/utils.py b/sky/skylet/providers/ibm/utils.py
new file mode 100644
index 000000000..7a5915ab1
--- /dev/null
+++ b/sky/skylet/providers/ibm/utils.py
@@ -0,0 +1,38 @@
+"""holds common utility function/constants to be used by the providers."""
+
+import logging
+import os
+import time
+
+RAY_RECYCLABLE = "ray-recyclable"
+
+
+def get_logger(caller_name):
+    """
+    Configures the logger of this module for console output and file output
+    logs of level DEBUG and higher will be directed to file under LOGS_FOLDER.
+    logs of level INFO and higher will be directed to console output.
+    """
+    logger = logging.getLogger(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)
+    logs_path = LOGS_FOLDER + caller_name + time.strftime("%Y-%m-%d--%H-%M-%S")
+    # pylint: disable=line-too-long
+    file_formatter = logging.Formatter(
+        "%(asctime)s %(levelname)-8s %(message)s", datefmt="%Y-%m-%d %H:%M:%S"
+    )
+
+    file_handler = logging.FileHandler(logs_path)
+    file_handler.setFormatter(file_formatter)
+    file_handler.setLevel(logging.DEBUG)
+
+    console_output_handler = logging.StreamHandler()
+    console_output_handler.setFormatter(file_formatter)
+    console_output_handler.setLevel(logging.INFO)
+
+    logger.addHandler(file_handler)
+    logger.addHandler(console_output_handler)
+    return logger
diff --git a/sky/skylet/providers/ibm/vpc_provider.py b/sky/skylet/providers/ibm/vpc_provider.py
new file mode 100644
index 000000000..ef6c9147b
--- /dev/null
+++ b/sky/skylet/providers/ibm/vpc_provider.py
@@ -0,0 +1,835 @@
+"""
+module allocating VPC - network namespace and configuration
+for Ray's cluster. used by the node_provider module to group the
+nodes under the same subnet, tagged by the same cluster name.
+"""
+
+from concurrent.futures import ThreadPoolExecutor
+import uuid
+import copy
+import time
+import requests
+import json
+import textwrap
+from sky.adaptors import ibm
+from sky.skylet.providers.ibm.utils import get_logger, RAY_RECYCLABLE
+
+# pylint: disable=line-too-long
+logger = get_logger("vpc_provider_")
+REQUIRED_RULES = {
+    "outbound_tcp_all": "selected security group is missing rule permitting outbound TCP access\n",
+    "outbound_udp_all": "selected security group is missing rule permitting outbound UDP access\n",
+    "inbound_tcp_sg": "selected security group is missing rule permitting inbound tcp traffic inside selected security group\n",
+    "inbound_tcp_22": "selected security group is missing rule permitting inbound traffic to tcp port 22 required for ssh\n",
+}
+INSECURE_RULES = {
+    "inbound_tcp_6379": "selected security group is missing rule permitting inbound traffic to tcp port 6379 required for Redis\n",
+    "inbound_tcp_8265": "selected security group is missing rule permitting inbound traffic to tcp port 8265 required to access Ray Dashboard\n",
+}
+REQUIRED_RULES.update(INSECURE_RULES)
+
+
+class IBMVPCProvider:
+    """
+    Manages a vpc containing the network configuration for the cluster's nodes.
+    """
+
+    def __init__(self, resource_group_id, region, cluster_name):
+        self.vpc_client = ibm.client(region=region)
+        self.search_client = ibm.search_client()
+        self.tagging_client = ibm.tagging_client()
+        self.resource_group_id = resource_group_id
+        self.cluster_name = cluster_name
+        ## region and zone might change between failovers
+        self.region = None
+        self.zone = None
+
+    def create_or_fetch_vpc(self, region, zone):
+        """
+        returns a cluster with tag matching the cluster name if exists, else creates one.
+        an entry point (out of 2) to this module.
+        """
+
+        # refresh client region scope if region changed.
+        if self.region and self.region != region:
+            self.vpc_client = ibm.client(region=region)
+        self.region = region
+        self.zone = zone
+        reused_vpc_data = None
+        # pylint: disable=line-too-long
+        vpcs_filtered_by_tags_and_region = self.search_client.search(
+            query=f"type:vpc AND tags:{self.cluster_name} AND region:{self.region}",
+            fields=["tags", "region", "type"],
+            limit=1000,
+        ).get_result()["items"]
+        for vpc in vpcs_filtered_by_tags_and_region:
+            vpc_id = vpc["crn"].rsplit(":", 1)[-1]
+            vpc_data = self.get_vpc_data(vpc_id, self.region)
+            if vpc_data["status"] == "available":
+                reused_vpc_data = vpc_data
+                break
+        # found vpc tagged with cluster name in the required region
+        if reused_vpc_data:
+            # using self.region since tagged vpc is in the same region
+            subnets = self.get_vpc_subnets(reused_vpc_data, self.region)
+            subnet_in_zone = next(
+                (subnet for subnet in subnets if subnet["zone"]["name"] == self.zone),
+                None,
+            )
+            # found a subnet in the required zone
+            if subnet_in_zone:
+                subnet_id = subnet_in_zone["id"]
+                public_gateway = subnet_in_zone.get("public_gateway")
+                if not public_gateway:
+                    public_gateway = self.create_public_gateway(
+                        reused_vpc_data["id"], self.zone, subnet_in_zone
+                    )
+            # tagged vpc found doesn't have a subnet in the required zone
+            else:
+                subnet_data = self.create_subnet(reused_vpc_data["id"], self.zone)
+                subnet_id = subnet_data["id"]
+                public_gateway = self.create_public_gateway(
+                    reused_vpc_data["id"], self.zone, subnet_data
+                )
+
+            # add missing security group rules if needed
+            security_group = reused_vpc_data.get("default_security_group")
+            if security_group:
+                sg_id = security_group["id"]
+                self.add_missing_sg_rules(sg_id)
+
+                # managed to reuse found VPC
+                logger.info(
+                    f"Reusing VPC {reused_vpc_data['id']} named: {reused_vpc_data['name']}"
+                )
+                return {
+                    "vpc_id": reused_vpc_data["id"],
+                    "subnet_id": subnet_id,
+                    "security_group_id": sg_id,
+                }
+
+        # delete a tagged vpc that doesn't meet requirements
+        if reused_vpc_data:
+            self.delete_vpc(reused_vpc_data["id"], self.region)
+        # create a new vpc
+        vpc_tags = self.create_vpc()
+        return vpc_tags
+
+    def create_vpc(self):
+        """creates a vpc, tags it and return key values pertaining it.
+
+        uuid is added to vpc name to avoid naming collision.
+        vpc is tagged using the exact cluster name, as appears on
+            the cluster's config file.
+
+        Returns:
+            dict: containing the keys: vpc_id, subnet_id
+                    and security_group_id
+        """
+        vpc_data = self.vpc_client.create_vpc(
+            address_prefix_management="auto",
+            classic_access=False,
+            name=f"sky-vpc-{self.cluster_name}-{str(uuid.uuid4())[:5]}",
+            resource_group={"id": self.resource_group_id},
+        ).get_result()
+        subnet_data = self.create_subnet(vpc_data["id"], self.zone)
+        self.create_public_gateway(vpc_data["id"], self.zone, subnet_data)
+        sg_id = self.create_sg_rules(vpc_data)
+
+        # tag vpc with the cluster's name
+        resource_model = {"resource_id": vpc_data["crn"]}
+        self.tagging_client.attach_tag(
+            resources=[resource_model], tag_names=[self.cluster_name], tag_type="user"
+        ).get_result()
+
+        return {
+            "vpc_id": vpc_data["id"],
+            "subnet_id": subnet_data["id"],
+            "security_group_id": sg_id,
+        }
+
+    def create_subnet(self, vpc_id, zone_name):
+        ipv4_cidr_block = None
+        subnet_name = f"sky-subnet-{self.cluster_name}-{str(uuid.uuid4())[:5]}"
+        res = self.vpc_client.list_vpc_address_prefixes(vpc_id).get_result()
+
+        # searching for the CIDR block (internal ip range) matching the
+        # specified zone of a VPC (whose region has already been set)
+        address_prefixes = res["address_prefixes"]
+        ipv4_cidr_block = next(
+            (
+                address_prefix["cidr"]
+                for address_prefix in address_prefixes
+                if address_prefix["zone"]["name"] == zone_name
+            ),
+            None,
+        )
+        if not ipv4_cidr_block:
+            raise Exception(
+                "Failed to locate a cidr block "
+                f"Matching the zone name: {zone_name} to create "
+                "a subnet"
+            )
+
+        subnet_prototype = {}
+        subnet_prototype["zone"] = {"name": zone_name}
+        subnet_prototype["ip_version"] = "ipv4"
+        subnet_prototype["name"] = subnet_name
+        subnet_prototype["resource_group"] = {"id": self.resource_group_id}
+        subnet_prototype["vpc"] = {"id": vpc_id}
+        subnet_prototype["ipv4_cidr_block"] = ipv4_cidr_block
+
+        subnet_data = self.vpc_client.create_subnet(subnet_prototype).get_result()
+        return subnet_data
+
+    def create_public_gateway(self, vpc_id, zone_name, subnet_data):
+
+        gateway_prototype = {}
+        gateway_prototype["vpc"] = {"id": vpc_id}
+        gateway_prototype["zone"] = {"name": zone_name}
+        gateway_prototype["name"] = f"{subnet_data['name']}-gw"
+        gateway_prototype["resource_group"] = {"id": self.resource_group_id}
+        gateway_data = self.vpc_client.create_public_gateway(
+            **gateway_prototype
+        ).get_result()
+        gateway_id = gateway_data["id"]
+
+        self.vpc_client.set_subnet_public_gateway(subnet_data["id"], {"id": gateway_id})
+        return gateway_id
+
+    def create_sg_rules(self, vpc_data):
+
+        sg_id = vpc_data["default_security_group"]["id"]
+        sg_name = f"{self.cluster_name}-{str(uuid.uuid4())[:5]}-sg"
+
+        # update sg name
+        self.vpc_client.update_security_group(
+            sg_id, security_group_patch={"name": sg_name}
+        )
+
+        # open private tcp traffic between VSIs within the security group
+        sg_rule_prototype = _build_security_group_rule_prototype_model(
+            "inbound_tcp_sg", sg_id=sg_id
+        )
+        self.vpc_client.create_security_group_rule(
+            sg_id, sg_rule_prototype
+        ).get_result()
+
+        # add all other required rules configured by the specific backend
+        for rule in REQUIRED_RULES.keys():
+            sg_rule_prototype = _build_security_group_rule_prototype_model(rule)
+            if sg_rule_prototype:
+                self.vpc_client.create_security_group_rule(
+                    sg_id, sg_rule_prototype
+                ).get_result()
+
+        return sg_id
+
+    def get_vpc_data(self, vpc_id, region):
+        """returns vpc data if exists, else None"""
+        if not vpc_id:
+            return None
+        tmp_vpc_client = ibm.client(region=region)
+        try:
+            vpc_data = tmp_vpc_client.get_vpc(vpc_id).result
+            return vpc_data
+        except ibm.ibm_cloud_sdk_core.ApiException as e:
+            if e.code == 404:
+                logger.debug("VPC doesn't exist.")
+                return None
+            else:
+                raise
+
+    def get_vpc_subnets(self, vpc_data, region, field=""):
+        """return data on subnets belonging to specified vpc within
+        the specified region.
+
+        if 'field' is specified narrowing data returned to
+        data['field'] (for each subnet)
+
+        Args:
+            vpc_data (str): vpc data as received from vpc client.
+            region (str): ibm vpc region.
+            field (str, optional): field within the subnet data response from vpc
+              client's list_subnets()/get_subnet() response. Defaults to ''.
+
+        Returns:
+            str: data on all subnets belonging to the specified vpc.
+        """
+        if not vpc_data:
+            return None
+        # pylint: disable=line-too-long
+        tmp_vpc_client = ibm.client(region=region)
+        subnets_attached_to_routing_table = tmp_vpc_client.list_subnets(
+            routing_table_id=vpc_data["default_routing_table"]["id"]
+        ).get_result()["subnets"]
+        if field:
+            return [subnet[field] for subnet in subnets_attached_to_routing_table]
+        else:
+            return subnets_attached_to_routing_table
+
+    def delete_vpc(self, vpc_id, region):
+        """
+        deletes a vpc with the specified id and region.
+        an entry point to this module (alongside create_or_fetch_vpc)
+        """
+        logger.debug(f"Deleting vpc: {vpc_id}")
+        tmp_vpc_client = ibm.client(region=region)
+        vpc_data = self.get_vpc_data(vpc_id, region)
+        if not vpc_data:
+            logger.warn(f"vpc:{vpc_id} is set for deletion, but wasn't found")
+            return None
+        self.delete_vms(tmp_vpc_client, vpc_id)
+        self.delete_subnets(tmp_vpc_client, vpc_data, region)
+        self.delete_gateways(tmp_vpc_client, vpc_id)
+        # at this point vpc was already verified to be existing
+        # thus no relevant exception to catch when deleting.
+        tmp_vpc_client.delete_vpc(vpc_id)
+
+    def delete_vms(self, vpc_client, vpc_id):
+        def _poll_vpc_contains_vms(vpc_id):
+            tries = 60
+            sleep_interval = 3
+            while tries:
+                # list_instances() never raise an exception, check values instead
+                res = vpc_client.list_instances(vpc_id=vpc_id).get_result()
+                if not res["total_count"]:
+                    return True
+                else:
+                    tries -= 1
+                    time.sleep(sleep_interval)
+            raise Exception(
+                "Failed to delete VPC's instances within "
+                "the expected time frame. Cannot "
+                "continue to delete VPC."
+            )
+
+        def _del_instance(vm_data):
+            # first delete ips created by node_provider
+            nic_id = vm_data["network_interfaces"][0]["id"]
+            res = vpc_client.list_instance_network_interface_floating_ips(
+                vm_data["id"], nic_id
+            ).get_result()
+            floating_ips = res.get("floating_ips", [])
+            for ip in floating_ips:
+                if ip["name"].startswith(RAY_RECYCLABLE):
+                    logger.debug(f"Deleting IP: {ip['id']}")
+                    vpc_client.delete_floating_ip(ip["id"])
+            logger.debug(f"Deleting VM: {vm_data['id']}")
+            vpc_client.delete_instance(id=vm_data["id"])
+
+        # pylint: disable=line-too-long E1136
+        res = vpc_client.list_instances(vpc_id=vpc_id).get_result()
+        num_instances = res["total_count"]
+
+        # Delete VSIs if exist
+        if num_instances:
+            instances = res["instances"]
+            with ThreadPoolExecutor(num_instances) as ex:
+                for i in range(num_instances):
+                    ex.submit(_del_instance, instances[i])
+            # wait until all vms are deleted to proceed
+            _poll_vpc_contains_vms(vpc_id)
+
+    def delete_subnets(self, vpc_client, vpc_data, region):
+        def _poll_subnet_deleted(subnet_id):
+            tries = 10
+            sleep_interval = 2
+            while tries:
+                try:
+                    vpc_client.get_subnet(subnet_id).get_result()
+                except ibm.ibm_cloud_sdk_core.ApiException:
+                    logger.debug(f"Deleted subnet id: {subnet_id}")
+                    return True
+                tries -= 1
+                time.sleep(sleep_interval)
+            logger.error("Failed to delete instance within the alloted time\n")
+            return False
+
+        for subnet_id in self.get_vpc_subnets(vpc_data, region, field="id"):
+            # get_result() used for synchronization
+            logger.debug(f"Deleting subnet: {subnet_id}")
+            vpc_client.delete_subnet(subnet_id).get_result()
+            _poll_subnet_deleted(subnet_id)
+
+    def delete_gateways(self, vpc_client, vpc_id):
+        """deletes all gateways attached to the specified vpc"""
+        # pylint: disable=line-too-long
+        gateways = vpc_client.list_public_gateways(
+            resource_group_id=self.resource_group_id
+        ).get_result()["public_gateways"]
+        gateways_ids_of_vpc = [
+            gateway["id"] for gateway in gateways if gateway["vpc"]["id"] == vpc_id
+        ]
+        for gateway_id in gateways_ids_of_vpc:
+            # get_result() used for synchronization
+            logger.debug(f"Deleting gateway: {gateway_id}")
+            vpc_client.delete_public_gateway(gateway_id).get_result()
+
+    def add_missing_sg_rules(self, sec_group_id):
+        missing_rules = self.get_unsatisfied_security_group_rules(sec_group_id)
+
+        if missing_rules:
+            for val in missing_rules.values():
+                logger.debug(f"missing {val}")
+            for missing_rule in missing_rules.keys():
+                sg_rule_prototype = _build_security_group_rule_prototype_model(
+                    missing_rule, sec_group_id
+                )
+                self.vpc_client.create_security_group_rule(
+                    sec_group_id, sg_rule_prototype
+                ).get_result()
+
+    def get_unsatisfied_security_group_rules(self, sg_id):
+        """
+        returns unsatisfied security group rules.
+        """
+
+        unsatisfied_rules = copy.deepcopy(REQUIRED_RULES)
+        sg = self.vpc_client.get_security_group(sg_id).get_result()
+
+        for rule in sg["rules"]:
+            # pylint: disable=line-too-long
+            # check outbound rules that are not associated with a specific IP address range
+            if rule["direction"] == "outbound" and rule["remote"] == {
+                "cidr_block": "0.0.0.0/0"
+            }:
+                if rule["protocol"] == "all":
+                    # outbound is fine!
+                    unsatisfied_rules.pop("outbound_tcp_all", None)
+                    unsatisfied_rules.pop("outbound_udp_all", None)
+                elif rule["protocol"] == "tcp":
+                    unsatisfied_rules.pop("outbound_tcp_all", None)
+                elif rule["protocol"] == "udp":
+                    unsatisfied_rules.pop("outbound_udp_all", None)
+
+            # Check inbound rules
+            elif rule["direction"] == "inbound":
+                # check rules that are not associated with a specific IP address range
+                if rule["remote"] == {"cidr_block": "0.0.0.0/0"}:
+                    # we interested only in all or tcp protocols
+                    if rule["protocol"] == "all":
+                        # there a rule permitting all traffic
+                        unsatisfied_rules.pop("inbound_tcp_sg", None)
+                        unsatisfied_rules.pop("inbound_tcp_22", None)
+                        unsatisfied_rules.pop("inbound_tcp_6379", None)
+                        unsatisfied_rules.pop("inbound_tcp_8265", None)
+
+                    elif rule["protocol"] == "tcp":
+                        if rule["port_min"] == 1 and rule["port_max"] == 65535:
+                            # all ports are open
+                            unsatisfied_rules.pop("inbound_tcp_sg", None)
+                            unsatisfied_rules.pop("inbound_tcp_22", None)
+                            unsatisfied_rules.pop("inbound_tcp_6379", None)
+                            unsatisfied_rules.pop("inbound_tcp_8265", None)
+                        else:
+                            port_min = rule["port_min"]
+                            port_max = rule["port_max"]
+                            if port_min <= 22 and port_max >= 22:
+                                unsatisfied_rules.pop("inbound_tcp_22", None)
+                            elif port_min <= 6379 and port_max >= 6379:
+                                unsatisfied_rules.pop("inbound_tcp_6379", None)
+                            elif port_min <= 8265 and port_max >= 8265:
+                                unsatisfied_rules.pop("inbound_tcp_8265", None)
+
+                # rule regards private traffic within the VSIs associated with the security group
+                elif rule["remote"].get("id") == sg["id"]:
+                    # validate that inbound traffic inside group available
+                    if rule["protocol"] == "all" or rule["protocol"] == "tcp":
+                        unsatisfied_rules.pop("inbound_tcp_sg", None)
+
+        return unsatisfied_rules
+
+    def remote_cluster_removal(self, vpc_id, region):
+        """deletes the vpc and its associated resources
+        using the FaaS service: ibm cloud functions.
+        Used only when cluster is set to be deleted
+        remotely, e.g. by `sky autostop --down`"""
+        cc = ClusterCleaner(self.resource_group_id, vpc_id, region)
+        cc.delete_cluster()
+
+
+def _build_security_group_rule_prototype_model(missing_rule, sg_id=None):
+    direction, protocol, port = missing_rule.split("_")
+    remote = {"cidr_block": "0.0.0.0/0"}
+
+    try:  # port number was specified
+        port = int(port)
+        port_min = port
+        port_max = port
+    # pylint: disable=W0703
+    except Exception:
+        port_min = 1
+        port_max = 65535
+
+        # only valid if security group already exists
+        if port == "sg":
+            if not sg_id:
+                return None
+            remote = {"id": sg_id}
+
+    return {
+        "direction": direction,
+        "ip_version": "ipv4",
+        "protocol": protocol,
+        "remote": remote,
+        "port_min": port_min,
+        "port_max": port_max,
+    }
+
+
+class ClusterCleaner:
+    """Responsible for deleting a cluster with all its associated
+    resources. Used when the remote cluster head is requested to
+    dismantle its cluster."""
+
+    # default region for the cloud function namespace
+    namespace_region = "us-east"
+    # default name for the cloud function namespace (cf doesn't support tagging)
+    namespace_name = "skypilot-namespace"
+    # default name for the cloud function action.
+    action_name = "skypilot-vpc-cleaner-action"
+    # url to cloud function's namespaces in the chosen region: `namespace_region`
+    cf_namespaces_url = (
+        f"https://{namespace_region}.functions.cloud.ibm.com/api/v1/namespaces"
+    )
+
+    def __init__(self, resource_group_id, vpc_id, vpc_region) -> None:
+        self.resource_group_id = resource_group_id
+        self.vpc_id = vpc_id
+        self.vpc_region = vpc_region
+
+    function_code = textwrap.dedent(
+        """
+    import subprocess
+    import time
+    from concurrent.futures import ThreadPoolExecutor
+    RAY_RECYCLABLE = "ray-recyclable"
+    ibm_vpc_client = None
+    # modules installed and imported entry point
+    ibm_vpc = None
+    ibm_cloud_sdk_core = None
+
+    def get_vpc_data(vpc_id):
+
+        if not vpc_id: return None
+        try:
+            vpc_data = ibm_vpc_client.get_vpc(vpc_id).result
+            return vpc_data
+        except ibm_cloud_sdk_core.ApiException as e:
+            if e.code == 404:
+                print(("VPC doesn't exist."))
+                return None
+            else: raise 
+
+    def delete_subnets(vpc_data):
+        def _poll_subnet_exists(subnet_id):
+            tries = 30 # waits up to 5 min with 10 sec interval
+            sleep_interval = 10
+            while tries:
+                try:
+                    subnet_data = ibm_vpc_client.get_subnet(subnet_id).result
+                except Exception:
+                    print('Deleted subnet id: {}'.format(subnet_id))
+                    return True
+                tries -= 1
+                time.sleep(sleep_interval)
+            print(f"Failed to delete instance within expected time frame of {tries*sleep_interval/60} minutes.")
+            return False
+
+        subnets_attached_to_routing_table = ibm_vpc_client.list_subnets(routing_table_id = vpc_data['default_routing_table']['id']).get_result()['subnets']
+        subnets_ids = [subnet['id'] for subnet in subnets_attached_to_routing_table]
+        for id in subnets_ids:
+            try:
+                ibm_vpc_client.delete_subnet(id).get_result()
+                _poll_subnet_exists(id)
+            except ibm_cloud_sdk_core.ApiException as e:
+                if e.code == 404:
+                    print("subnet doesn't exist.")
+
+    def delete_gateways(vpc_id):
+        gateways = ibm_vpc_client.list_public_gateways(resource_group_id=RESOURCE_GROUP_ID).get_result()['public_gateways']
+        gateways_ids_of_vpc = [gateway['id'] for gateway in gateways if gateway['vpc']['id']== vpc_id]
+        for gateway_id in gateways_ids_of_vpc:
+            deleting_resource = True
+            while deleting_resource:
+                try:
+                    ibm_vpc_client.delete_public_gateway(gateway_id).get_result()
+                    deleting_resource = False
+                except ibm_cloud_sdk_core.ApiException as e:
+                    if e.code == 404:
+                        print("gateway doesn't exist.") 
+                        deleting_resource = False
+                    if e.code == 409:
+                        print("gateway still in use.")
+                        # will retry until cloud functions timeout. 
+                        time.sleep(5) 
+
+    def delete_vms(vpc_id):
+        def _poll_vpc_contains_vms(vpc_id):
+            tries = 60
+            sleep_interval = 3
+            while tries:
+                # list_instances() never raise an exception, check values instead
+                res = ibm_vpc_client.list_instances(vpc_id=vpc_id).get_result()
+                if not res["total_count"]:
+                    return True
+                else:
+                    tries -= 1
+                    time.sleep(sleep_interval)
+            raise Exception(
+                "Failed to delete VPC's instances within "
+                "the expected time frame. Cannot "
+                "continue to delete VPC."
+            )
+
+        def _del_instance(vm_data):
+            # first delete ips created by node_provider 
+            nic_id = vm_data["network_interfaces"][0]["id"]
+            res = ibm_vpc_client.list_instance_network_interface_floating_ips(
+                vm_data["id"], nic_id
+            ).get_result()
+            floating_ips = res.get("floating_ips", [])
+            for ip in floating_ips:
+                if ip["name"].startswith(RAY_RECYCLABLE):
+                    print(f"Deleting IP: {ip['id']}")
+                    ibm_vpc_client.delete_floating_ip(ip["id"])
+            print(f"Deleting VM: {vm_data['id']}")
+            ibm_vpc_client.delete_instance(id=vm_data["id"])
+            
+        res = ibm_vpc_client.list_instances(vpc_id=vpc_id).get_result()
+        num_instances = res["total_count"]
+
+        # Delete VSIs if exist
+        if num_instances:
+            instances = res["instances"]
+            with ThreadPoolExecutor(num_instances) as ex:
+                for i in range(num_instances):
+                    ex.submit(_del_instance, instances[i])
+            # wait until all vms are deleted to proceed
+            _poll_vpc_contains_vms(vpc_id)
+
+    def delete_unbound_vpc(vpc_id):
+        deleting_resource = True
+        while deleting_resource:
+            try:
+                ibm_vpc_client.delete_vpc(vpc_id).get_result()
+                deleting_resource = False
+            except ibm_cloud_sdk_core.ApiException as e:
+                if e.code == 404:
+                    print("VPC doesn't exist.") 
+                    deleting_resource = False
+                if e.code == 409:
+                    print("VPC still in use.")
+                    # will retry until cloud functions timeout. 
+                    time.sleep(5) 
+
+    def delete_vpc(vpc_id):
+        vpc_data = get_vpc_data(vpc_id)
+        if not vpc_data:
+            print((f"Failed to find a VPC with id={vpc_id}"))
+            return
+        print(f"Deleting vpc:{vpc_data['name']} with id:{vpc_id}")
+        delete_vms(vpc_id)
+        delete_subnets(vpc_data)
+        delete_gateways(vpc_id)
+        delete_unbound_vpc(vpc_id)
+        print(f"VPC {vpc_data['name']} and its attached resources were deleted successfully")
+
+
+    def main(dict):
+        global ibm_vpc_client, RESOURCE_GROUP_ID, ibm_cloud_sdk_core, ibm_vpc
+        def install_package(package):
+            pip_location_stdout = subprocess.run(['which', 'pip'], capture_output=True, text=True)
+            pip_location = pip_location_stdout.stdout.strip()
+            subprocess.call([pip_location, 'install', package], stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)
+
+        for package in ["ibm-vpc", "ibm-cloud-sdk-core"]:
+            install_package(package)
+
+        import ibm_vpc as _ibm_vpc
+        import ibm_cloud_sdk_core as _ibm_cloud_sdk_core
+        ibm_vpc = _ibm_vpc
+        ibm_cloud_sdk_core = _ibm_cloud_sdk_core
+
+        iam_api_key, RESOURCE_GROUP_ID, vpc_id, region = dict['iam_api_key'], dict['resource_group_id'], dict['vpc_id'], dict['region']
+
+        authenticator = ibm_cloud_sdk_core.authenticators.IAMAuthenticator(iam_api_key, url=None)
+        ibm_vpc_client = ibm_vpc.VpcV1('2022-06-30',authenticator=authenticator)
+
+        if not region:
+            raise Exception("VPC not found in any region")
+
+        ibm_vpc_client.set_service_url(f'https://{region}.iaas.cloud.ibm.com/v1')
+
+        delete_vpc(vpc_id=vpc_id)
+        return {"Status": "Success"}
+    """
+    )
+
+    def get_headers(self):
+        return {
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+            "Authorization": "Bearer " + ibm.get_oauth_token(),
+        }
+
+    def create_or_fetch_namespace(self):
+        """returns namespace id.
+        creates a namespace with given name in specified region.
+        if namespace exists returns its id instead."""
+
+        def _create_new_namespace():
+            logger.info(
+                f"Creating a new namespace: {self.namespace_name} in {self.namespace_region}"
+            )
+
+            data = {
+                "name": self.namespace_name,
+                "resource_group_id": self.resource_group_id,
+                "resource_plan_id": "functions-base-plan",
+            }
+
+            res = requests.post(
+                self.cf_namespaces_url, headers=self.get_headers(), json=data
+            ).json()
+            if res.status_code != 200:
+                logger.error(res.text)
+            namespace_id = res["id"]
+            logger.info(f"Created new namespace with id: {namespace_id}")
+            return namespace_id
+
+        def _get_cloud_function_namespaces_metadata(offset=0):
+            """returns meta data on namespaces of ibm cloud functions within a specified region
+            :param offset - offset from the beginning of the list of results attained from the GET request,
+                            which may contain up to 200 namespaces per http response"""
+
+            res = requests.get(
+                f"{self.cf_namespaces_url}?limit=200&offset={offset}",
+                headers=self.get_headers(),
+            )
+            return json.loads(res.text)
+
+        def _get_cloud_function_namespaces():
+            """returns relevant metadata on existing namespaces within a given region."""
+            logger.info(
+                f"Obtaining Cloud Function namespaces in {self.namespace_region}"
+            )
+
+            namespaces = []
+
+            collecting_namespaces = True
+            max_limit = 200
+            offset = 0
+
+            #  request for namespaces is limited to 200 at a time, thus the request is fulfilled in increments of 200s.
+            while collecting_namespaces:
+                namespace_metadata = _get_cloud_function_namespaces_metadata(offset)
+                if namespace_metadata["total_count"] == max_limit:
+                    offset += max_limit
+                else:
+                    collecting_namespaces = False
+
+                for name_space in namespace_metadata["namespaces"]:
+                    if "name" in name_space:  # API based namespace
+                        namespaces.append(
+                            {
+                                "name": name_space["name"],
+                                "type": "API_based",
+                                "id": name_space["id"],
+                                "region": name_space["location"],
+                            }
+                        )
+
+                    else:  # cloud foundry based namespace
+                        namespaces.append(
+                            {
+                                "name": name_space["id"],
+                                "type": "CF_based",
+                                "region": name_space["location"],
+                            }
+                        )
+
+            return namespaces
+
+        namespaces_in_region = _get_cloud_function_namespaces()
+        target_namespace_id = None
+        if namespaces_in_region:
+            target_namespace_id = next(
+                (
+                    namespace["id"]
+                    for namespace in namespaces_in_region
+                    if namespace["name"] == self.namespace_name
+                ),
+                None,
+            )
+        if not target_namespace_id:
+            target_namespace_id = _create_new_namespace()
+        else:
+            logger.info(f"Reusing namespace: {target_namespace_id}")
+        return target_namespace_id
+
+    def _get_cloud_functions_actions(self, namespace_id):
+        """returns meta data on namespaces of ibm cloud functions within a specified region
+        :param offset - offset from the beginning of the list of results attained from the GET request,
+                        which may contain up to 200 namespaces per http response"""
+
+        res = requests.get(
+            f"{self.cf_namespaces_url}/{namespace_id}/actions?limit=200",
+            headers=self.get_headers(),
+        )
+        return json.loads(res.text)
+
+    def create_action(self, namespace_id):
+        logger.info(f"creating action on namespace: {namespace_id}")
+        # Define the function parameters
+        function_params = {
+            "exec": {"kind": "python:3.9", "code": self.function_code},
+            "limits": {"timeout": 600000},
+        }
+        res = requests.put(
+            f"{self.cf_namespaces_url}/{namespace_id}/actions/{self.action_name}?blocking=true&overwrite=true",
+            headers=self.get_headers(),
+            data=json.dumps(function_params),
+        )
+        if res.status_code != 200:
+            logger.error(res.text)
+        return json.loads(res.text)
+
+    def delete_action(self, namespace_id):
+        """return the deleted function's metadata if it existed."""
+        logger.info(f"deleting action on namespace: {namespace_id}")
+        res = requests.delete(
+            f"{self.cf_namespaces_url}/{namespace_id}/actions/{self.action_name}?blocking=true",
+            headers=self.get_headers(),
+        )
+        if res.status_code != 200:
+            logger.warn(res.text)
+        return json.loads(res.text)
+
+    def invoke_action(self, namespace_id: str):
+        logger.info(f"invoking action on namespace: {namespace_id}")
+        payload = {
+            "iam_api_key": ibm.get_api_key(),
+            "resource_group_id": self.resource_group_id,
+            "vpc_id": self.vpc_id,
+            "region": self.vpc_region,
+        }
+        res = requests.post(
+            f"{self.cf_namespaces_url}/{namespace_id}/actions/{self.action_name}?blocking=true",
+            headers=self.get_headers(),
+            data=json.dumps(payload),
+        )
+        if res.status_code != 200:
+            logger.error(res.text)
+        return json.loads(res.text)
+
+    def delete_cluster(self):
+        """Deletes VPC with id==self.vpc_id.
+        1. creates a CloudFunctions namespace named ClusterCleaner.namespace_name if doesn't exists.
+        2. using idempotent function that deletes an action named ClusterCleaner.action_name if exists.
+        3. invokes the action to delete the VPC and all its resources."""
+        cf_namespace_id = self.create_or_fetch_namespace()
+        self.delete_action(cf_namespace_id)
+        self.create_action(cf_namespace_id)
+        self.invoke_action(cf_namespace_id)
diff --git a/sky/templates/ibm-ray.yml.j2 b/sky/templates/ibm-ray.yml.j2
new file mode 100644
index 000000000..1790161e4
--- /dev/null
+++ b/sky/templates/ibm-ray.yml.j2
@@ -0,0 +1,148 @@
+cluster_name: {{cluster_name}}
+
+# The maximum number of workers nodes to launch in addition to the head node.
+max_workers: {{num_nodes - 1}}
+upscaling_speed: {{num_nodes - 1}}
+idle_timeout_minutes: 60
+
+
+provider:
+  type: external
+  module: sky.skylet.providers.ibm.IBMVPCNodeProvider
+  region: {{region}}
+  availability_zone: {{zones}}
+  # Keep (otherwise cannot reuse when re-provisioning).
+  # teardown(terminate=True) will override this.
+  cache_stopped_nodes: True
+  iam_api_key: {{iam_api_key}}
+  resource_group_id: {{resource_group_id}}
+  # Disable launch config check for worker nodes as it can cause resource
+  # leakage.
+  # Reference: https://github.com/ray-project/ray/blob/840215bc09e942b50cad0ab2db96a8fdc79217c1/python/ray/autoscaler/_private/autoscaler.py#L1101
+  # The upper-level SkyPilot code has make sure there will not be resource
+  # leakage.
+  disable_launch_config_check: true
+
+auth:
+  ssh_user: ubuntu
+  ssh_private_key: {{ssh_private_key}}
+
+
+available_node_types:
+  ray_head_default:
+    resources: {{instance_resources}}
+    node_config:
+      image_id: {{image_id}}  
+      boot_volume_capacity: {{disk_capacity}}
+      volume_tier_name: general-purpose
+      instance_profile_name: {{instance_type}}
+
+      {% if use_spot %}
+      InstanceMarketOptions:
+          MarketType: spot
+
+      {% endif %}
+{% if num_nodes > 1 %}
+  ray_worker_default:
+    min_workers: {{num_nodes - 1}}
+    max_workers: {{num_nodes - 1}}
+    resources: {{worker_instance_resources}}
+    node_config:
+      image_id: {{image_id}} 
+      boot_volume_capacity: {{disk_capacity}}
+      volume_tier_name: general-purpose
+      instance_profile_name: {{worker_instance_type}}
+
+      {% if use_spot %}
+      InstanceMarketOptions:
+          MarketType: spot
+
+      {% endif %}
+{%- endif %}
+
+head_node_type: ray_head_default
+
+
+{# Format: `REMOTE_PATH : LOCAL_PATH` #}
+file_mounts: {
+  "{{sky_ray_yaml_remote_path}}": "{{sky_ray_yaml_local_path}}",
+  "{{sky_remote_path}}/{{sky_wheel_hash}}": "{{sky_local_path}}",
+{%- for remote_path, local_path in credentials.items() %}
+  "{{remote_path}}": "{{local_path}}",
+{%- endfor %}
+}
+
+rsync_exclude: []
+
+initialization_commands: []
+
+# List of shell commands to run to set up nodes.
+# NOTE: these are very performance-sensitive. Each new item opens/closes an SSH
+# connection, which is expensive. Try your best to co-locate commands into fewer
+# items!
+#
+# Increment the following for catching performance bugs easier:
+#   current num items (num SSH connections): 1
+setup_commands:
+  # Create ~/.ssh/config file in case the file does not exist in the custom image.
+  # Make sure python3 & pip3 are available on this image.
+  # We set auto_activate_base to be false for pre-installed conda.
+  # This also kills the service that is holding the lock on dpkg (problem only exists on aws/azure, not gcp)
+  # Line 'sudo bash ..': set the ulimit as suggested by ray docs for performance. https://docs.ray.io/en/latest/cluster/vms/user-guides/large-cluster-best-practices.html#system-configuration
+  # Line 'sudo grep ..': set the number of threads per process to unlimited to avoid ray job submit stucking issue when the number of running ray jobs increase.
+  # Line 'mkdir -p ..': disable host key check
+  # Line 'python3 -c ..': patch the buggy ray files and enable `-o allow_other` option for `goofys`
+  - sudo systemctl stop unattended-upgrades || true;
+    sudo systemctl disable unattended-upgrades || true;
+    sudo sed -i 's/Unattended-Upgrade "1"/Unattended-Upgrade "0"/g' /etc/apt/apt.conf.d/20auto-upgrades || true;
+    sudo kill -9 `sudo lsof /var/lib/dpkg/lock-frontend | awk '{print $2}' | tail -n 1` || true;
+    sudo pkill -9 apt-get;
+    sudo pkill -9 dpkg;
+    sudo dpkg --configure -a;
+    mkdir -p ~/.ssh; touch ~/.ssh/config;
+    pip3 --version > /dev/null 2>&1 || (curl -sSL https://bootstrap.pypa.io/get-pip.py -o get-pip.py && python3 get-pip.py && echo "PATH=$HOME/.local/bin:$PATH" >> ~/.bashrc);
+    (type -a python | grep -q python3) || echo 'alias python=python3' >> ~/.bashrc;
+    (type -a pip | grep -q pip3) || echo 'alias pip=pip3' >> ~/.bashrc;
+    (which conda > /dev/null 2>&1 && conda init > /dev/null && conda config --set auto_activate_base false) || (wget -nc https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh && bash Miniconda3-latest-Linux-x86_64.sh -b && eval "$(~/miniconda3/bin/conda shell.bash hook)" && conda init && conda config --set auto_activate_base true);
+    source ~/.bashrc;
+    (pip3 list | grep ray | grep {{ray_version}} 2>&1 > /dev/null || pip3 install -U ray[default]=={{ray_version}}) && mkdir -p ~/sky_workdir && mkdir -p ~/.sky/sky_app;
+    (pip3 list | grep skypilot && [ "$(cat {{sky_remote_path}}/current_sky_wheel_hash)" == "{{sky_wheel_hash}}" ]) || (pip3 uninstall skypilot -y; pip3 install "$(echo {{sky_remote_path}}/{{sky_wheel_hash}}/skypilot-{{sky_version}}*.whl)[ibm]" && echo "{{sky_wheel_hash}}" > {{sky_remote_path}}/current_sky_wheel_hash || exit 1);
+    sudo bash -c 'rm -rf /etc/security/limits.d; echo "* soft nofile 1048576" >> /etc/security/limits.conf; echo "* hard nofile 1048576" >> /etc/security/limits.conf';
+    sudo grep -e '^DefaultTasksMax' /etc/systemd/system.conf || (sudo bash -c 'echo "DefaultTasksMax=infinity" >> /etc/systemd/system.conf'); sudo systemctl set-property user-$(id -u $(whoami)).slice TasksMax=infinity; sudo systemctl daemon-reload;
+    mkdir -p ~/.ssh; (grep -Pzo -q "Host \*\n  StrictHostKeyChecking no" ~/.ssh/config) || printf "Host *\n  StrictHostKeyChecking no\n" >> ~/.ssh/config;
+    python3 -c "from sky.skylet.ray_patches import patch; patch()" || exit 1;
+    [ -f /etc/fuse.conf ] && sudo sed -i 's/#user_allow_other/user_allow_other/g' /etc/fuse.conf || (sudo sh -c 'echo "user_allow_other" > /etc/fuse.conf'); # This is needed for `-o allow_other` option for `goofys`;
+
+
+# Command to start ray on the head node. You don't need to change this.
+# NOTE: these are very performance-sensitive. Each new item opens/closes an SSH
+# connection, which is expensive. Try your best to co-locate commands into fewer
+# items! The same comment applies for worker_start_ray_commands.
+#
+# Increment the following for catching performance bugs easier:
+#   current num items (num SSH connections): 1
+head_start_ray_commands:
+  # Start skylet daemon. (Should not place it in the head_setup_commands, otherwise it will run before sky is installed.)
+  # NOTE: --disable-usage-stats in `ray start` saves 10 seconds of idle wait.
+  # Line "which prlimit ..": increase the limit of the number of open files for the raylet process, as the `ulimit` may not take effect at this point, because it requires
+  # all the sessions to be reloaded. This is a workaround.
+  - ((ps aux | grep -v nohup | grep -v grep | grep -q -- "python3 -m sky.skylet.skylet") || nohup python3 -m sky.skylet.skylet >> ~/.sky/skylet.log 2>&1 &);
+    ray stop; RAY_SCHEDULER_EVENTS=0 ray start --disable-usage-stats --head --port=6379 --object-manager-port=8076 --autoscaling-config=~/ray_bootstrap_config.yaml {{"--resources='%s'" % custom_resources if custom_resources}} || exit 1;
+    which prlimit && for id in $(pgrep -f raylet/raylet); do sudo prlimit --nofile=1048576:1048576 --pid=$id || true; done;
+
+{%- if num_nodes > 1 %}
+worker_start_ray_commands:
+  - ray stop; RAY_SCHEDULER_EVENTS=0 ray start --disable-usage-stats --address=$RAY_HEAD_IP:6379 --object-manager-port=8076 {{"--resources='%s'" % custom_resources if custom_resources}} || exit 1;
+    which prlimit && for id in $(pgrep -f raylet/raylet); do sudo prlimit --nofile=1048576:1048576 --pid=$id || true; done;
+{%- else %}
+worker_start_ray_commands: []
+{%- endif %}
+
+head_node: {}
+worker_nodes: {}
+
+# These fields are required for external cloud providers.
+head_setup_commands: []
+worker_setup_commands: []
+cluster_synced_files: []
+file_mounts_sync_continuously: False


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 765fd9762..dbc919a46 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -18,7 +18,9 @@ from typing import List
 # --aws, --gcp, --azure, or --lambda.
 #
 # To only run tests for managed spot (without generic tests), use --managed-spot.
-all_clouds_in_smoke_tests = ['aws', 'gcp', 'azure', 'lambda', 'cloudflare']
+all_clouds_in_smoke_tests = [
+    'aws', 'gcp', 'azure', 'lambda', 'cloudflare', 'ibm'
+]
 default_clouds_to_run = ['gcp', 'azure']
 
 # Translate cloud name to pytest keyword. We need this because
@@ -29,7 +31,8 @@ cloud_to_pytest_keyword = {
     'gcp': 'gcp',
     'azure': 'azure',
     'lambda': 'lambda_cloud',
-    'cloudflare': 'cloudflare'
+    'cloudflare': 'cloudflare',
+    'ibm': 'ibm',
 }
 
 
diff --git a/tests/test_optimizer_dryruns.py b/tests/test_optimizer_dryruns.py
index f36c8a09e..2d288b2c9 100644
--- a/tests/test_optimizer_dryruns.py
+++ b/tests/test_optimizer_dryruns.py
@@ -437,7 +437,7 @@ def test_invalid_image(monkeypatch):
 
     with pytest.raises(ValueError) as e:
         _test_resources(monkeypatch, cloud=sky.Azure(), image_id='some-image')
-    assert 'only supported for AWS and GCP' in str(e.value)
+    assert 'only supported for AWS, GCP and IBM' in str(e.value)
 
 
 def test_valid_image(monkeypatch):
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index 130849a61..2f9cf3617 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -240,6 +240,23 @@ def test_gcp_region():
     run_one_test(test)
 
 
+@pytest.mark.ibm
+def test_ibm_region():
+    name = _get_cluster_name()
+    region = 'eu-de'
+    test = Test(
+        'region',
+        [
+            f'sky launch -y -c {name} --cloud ibm --region {region} examples/minimal.yaml',
+            f'sky exec {name} --cloud ibm examples/minimal.yaml',
+            f'sky logs {name} 1 --status',  # Ensure the job succeeded.
+            f'sky status --all | grep {name} | grep {region}',  # Ensure the region is correct.
+        ],
+        f'sky down -y {name}',
+    )
+    run_one_test(test)
+
+
 @pytest.mark.azure
 def test_azure_region():
     name = _get_cluster_name()
@@ -273,6 +290,23 @@ def test_aws_zone():
     run_one_test(test)
 
 
+@pytest.mark.ibm
+def test_ibm_zone():
+    name = _get_cluster_name()
+    zone = 'eu-de-2'
+    test = Test(
+        'zone',
+        [
+            f'sky launch -y -c {name} --cloud ibm examples/minimal.yaml --zone {zone}',
+            f'sky exec {name} --cloud ibm examples/minimal.yaml --zone {zone}',
+            f'sky logs {name} 1 --status',  # Ensure the job succeeded.
+            f'sky status --all | grep {name} | grep {zone}',  # Ensure the zone is correct.
+        ],
+        f'sky down -y {name} {name}-2 {name}-3',
+    )
+    run_one_test(test)
+
+
 @pytest.mark.gcp
 def test_gcp_zone():
     name = _get_cluster_name()
@@ -734,6 +768,7 @@ def test_cli_logs(generic_cloud: str):
 
 # ---------- Job Queue. ----------
 @pytest.mark.no_lambda_cloud  # Lambda Cloud does not have K80 gpus
+@pytest.mark.no_ibm  # IBM Cloud does not have K80 gpus. run test_ibm_job_queue instead
 def test_job_queue(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -783,7 +818,31 @@ def test_lambda_job_queue():
     run_one_test(test)
 
 
+@pytest.mark.ibm
+def test_ibm_job_queue():
+    name = _get_cluster_name()
+    test = Test(
+        'ibm_job_queue',
+        [
+            f'sky launch -y -c {name} --cloud ibm --gpus v100',
+            f'sky exec {name} -n {name}-1 --cloud ibm -d examples/job_queue/job_ibm.yaml',
+            f'sky exec {name} -n {name}-2 --cloud ibm -d examples/job_queue/job_ibm.yaml',
+            f'sky exec {name} -n {name}-3 --cloud ibm -d examples/job_queue/job_ibm.yaml',
+            f'sky queue {name} | grep {name}-1 | grep RUNNING',
+            f'sky queue {name} | grep {name}-2 | grep RUNNING',
+            f'sky queue {name} | grep {name}-3 | grep PENDING',
+            f'sky cancel -y {name} 2',
+            'sleep 5',
+            f'sky queue {name} | grep {name}-3 | grep RUNNING',
+            f'sky cancel -y {name} 3',
+        ],
+        f'sky down -y {name}',
+    )
+    run_one_test(test)
+
+
 @pytest.mark.no_lambda_cloud  # Lambda Cloud does not have T4 gpus
+@pytest.mark.no_ibm  # IBM Cloud does not have T4 gpus. run test_ibm_job_queue_multinode instead
 def test_job_queue_multinode(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -840,8 +899,47 @@ def test_large_job_queue(generic_cloud: str):
     run_one_test(test)
 
 
+@pytest.mark.ibm
+def test_ibm_job_queue_multinode():
+    name = _get_cluster_name()
+    task_file = 'examples/job_queue/job_multinode_ibm.yaml'
+    test = Test(
+        'ibm_job_queue_multinode',
+        [
+            f'sky launch -y -c {name} --cloud ibm --gpus v100 --num-nodes 2',
+            f'sky exec {name} -n {name}-1 -d {task_file}',
+            f'sky exec {name} -n {name}-2 -d {task_file}',
+            f'sky launch -y -c {name} -n {name}-3 --detach-setup -d {task_file}',
+            f's=$(sky queue {name}) && printf "$s" && (echo "$s" | grep {name}-1 | grep RUNNING)',
+            f's=$(sky queue {name}) && printf "$s" && (echo "$s" | grep {name}-2 | grep RUNNING)',
+            f's=$(sky queue {name}) && printf "$s" && (echo "$s" | grep {name}-3 | grep SETTING_UP)',
+            'sleep 90',
+            f's=$(sky queue {name}) && printf "$s" && (echo "$s" | grep {name}-3 | grep PENDING)',
+            f'sky cancel -y {name} 1',
+            'sleep 5',
+            f'sky queue {name} | grep {name}-3 | grep RUNNING',
+            f'sky cancel -y {name} 1 2 3',
+            f'sky launch -c {name} -n {name}-4 --detach-setup -d {task_file}',
+            # Test the job status is correctly set to SETTING_UP, during the setup is running,
+            # and the job can be cancelled during the setup.
+            f's=$(sky queue {name}) && printf "$s" && (echo "$s" | grep {name}-4 | grep SETTING_UP)',
+            f'sky cancel -y {name} 4',
+            f's=$(sky queue {name}) && printf "$s" && (echo "$s" | grep {name}-4 | grep CANCELLED)',
+            f'sky exec {name} --gpus v100:0.2 "[[ \$SKYPILOT_NUM_GPUS_PER_NODE -eq 1 ]] || exit 1"',
+            f'sky exec {name} --gpus v100:0.2 --num-nodes 2 "[[ \$SKYPILOT_NUM_GPUS_PER_NODE -eq 1 ]] || exit 1"',
+            f'sky exec {name} --gpus v100:1 --num-nodes 2 "[[ \$SKYPILOT_NUM_GPUS_PER_NODE -eq 1 ]] || exit 1"',
+            f'sky logs {name} 5 --status',
+            f'sky logs {name} 6 --status',
+            f'sky logs {name} 7 --status',
+        ],
+        f'sky down -y {name}',
+    )
+    run_one_test(test)
+
+
 # ---------- Submitting multiple tasks to the same cluster. ----------
 @pytest.mark.no_lambda_cloud  # Lambda Cloud does not have K80 gpus
+@pytest.mark.no_ibm  # IBM Cloud does not have K80 gpus
 def test_multi_echo(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -863,6 +961,7 @@ def test_multi_echo(generic_cloud: str):
 
 # ---------- Task: 1 node training. ----------
 @pytest.mark.no_lambda_cloud  # Lambda Cloud does not have V100 gpus
+@pytest.mark.no_ibm  # IBM cloud currently doesn't provide public image with CUDA
 def test_huggingface(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -972,6 +1071,7 @@ def test_multi_hostname(generic_cloud: str):
 
 # ---------- Task: n=2 nodes with setups. ----------
 @pytest.mark.no_lambda_cloud  # Lambda Cloud does not have V100 gpus
+@pytest.mark.no_ibm  # IBM cloud currently doesn't provide public image with CUDA
 def test_distributed_tf(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -1036,6 +1136,7 @@ def test_azure_start_stop():
 
 # ---------- Testing Autostopping ----------
 @pytest.mark.no_lambda_cloud  # Lambda Cloud does not support stopping instances
+@pytest.mark.no_ibm  # FIX(IBM) sporadically fails, as restarted workers stay uninitialized indefinitely
 def test_autostop(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -1169,6 +1270,7 @@ def test_cancel_azure():
 
 
 @pytest.mark.no_lambda_cloud  # Lambda Cloud does not have V100 gpus
+@pytest.mark.no_ibm  # IBM cloud currently doesn't provide public image with CUDA
 def test_cancel_pytorch(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -1191,8 +1293,30 @@ def test_cancel_pytorch(generic_cloud: str):
     run_one_test(test)
 
 
+# can't use `_get_cancel_task_with_cloud()`, as command `nvidia-smi`
+# requires a CUDA public image, which IBM doesn't offer
+@pytest.mark.ibm
+def test_cancel_ibm():
+    name = _get_cluster_name()
+    test = Test(
+        'ibm-cancel-task',
+        [
+            f'sky launch -y -c {name} --cloud ibm examples/minimal.yaml',
+            f'sky exec {name} -n {name}-1 -d  "while true; do echo \'Hello SkyPilot\'; sleep 2; done"',
+            'sleep 20',
+            f'sky queue {name} | grep {name}-1 | grep RUNNING',
+            f'sky cancel -y {name} 2',
+            f'sleep 5',
+            f'sky queue {name} | grep {name}-1 | grep CANCELLED',
+        ],
+        f'sky down -y {name}',
+    )
+    run_one_test(test)
+
+
 # ---------- Testing use-spot option ----------
 @pytest.mark.no_lambda_cloud  # Lambda Cloud does not support spot instances
+@pytest.mark.no_ibm  # IBM Cloud does not support spot instances
 def test_use_spot(generic_cloud: str):
     """Test use-spot and sky exec."""
     name = _get_cluster_name()
@@ -1211,6 +1335,7 @@ def test_use_spot(generic_cloud: str):
 
 # ---------- Testing managed spot ----------
 @pytest.mark.no_lambda_cloud  # Lambda Cloud does not support spot instances
+@pytest.mark.no_ibm  # IBM Cloud does not support spot instances
 @pytest.mark.managed_spot
 def test_spot(generic_cloud: str):
     """Test the spot yaml."""
@@ -1241,6 +1366,7 @@ def test_spot(generic_cloud: str):
 
 
 @pytest.mark.no_lambda_cloud  # Lambda Cloud does not support spot instances
+@pytest.mark.no_ibm  # IBM Cloud does not support spot instances
 @pytest.mark.managed_spot
 def test_spot_failed_setup(generic_cloud: str):
     """Test managed spot job with failed setup."""
@@ -1325,6 +1451,7 @@ def test_spot_recovery_gcp():
 
 
 @pytest.mark.no_lambda_cloud  # Lambda Cloud does not support spot instances
+@pytest.mark.no_ibm  # IBM Cloud does not support spot instances
 @pytest.mark.managed_spot
 def test_spot_recovery_default_resources(generic_cloud: str):
     """Test managed spot recovery for default resources."""
@@ -1529,6 +1656,7 @@ def test_spot_cancellation_gcp():
 
 # ---------- Testing storage for managed spot ----------
 @pytest.mark.no_lambda_cloud  # Lambda Cloud does not support spot instances
+@pytest.mark.no_ibm  # IBM Cloud does not support spot instances
 @pytest.mark.managed_spot
 def test_spot_storage(generic_cloud: str):
     """Test storage with managed spot"""
@@ -1581,6 +1709,7 @@ def test_spot_tpu():
 
 # ---------- Testing env for spot ----------
 @pytest.mark.no_lambda_cloud  # Lambda Cloud does not support spot instances
+@pytest.mark.no_ibm  # IBM Cloud does not support spot instances
 @pytest.mark.managed_spot
 def test_spot_inline_env(generic_cloud: str):
     """Test spot env"""