Repository: skypilot-org/skypilot
Pull request #1909 (2023-04-28 03:43:12)

Oci support
<!-- Describe the changes in this PR -->
This PR is the first implementation version of the Oracle Cloud (OCI) support in SkyPilot. So far as this notes writing, The supports including OCI CPU VM shapes, GPU VM shapes, CPU BM shapes, GPU BM shapes. It also supports NVIDIA GPU Cloud Machine Image. In the future, more functions/extensions will be added, prioritized by customer requirements.  


<!-- Describe the tests ran -->
<!-- Unit tests (tests/test_*.py) are part of GitHub CI; below are tests that launch on the cloud. -->

Tested (run the relevant ones):

- [ ] Any manual or new tests for this PR (please specify below)
      - sky launch with region/zone/instance_type/cpu/memory/image/use_spot parameters
      - sky launch without instance_type specified (inference of instance_type)
      - sky launch with image tag specified (map the tag to the id according to images.csv file)
      - sky launch with use_spot true/false (true=preemptible instances, on-demand type otherwise)
      - sky launch with cpu but no memory specified
      - sky launch without cpu & memory specified (create with default shape [2vCPUs * 16G Memory])
      - sky launch with CPU instances
      - sky launch with VM A10 GPU instance
      - sky launch with NVIDIA GPU Cloud Machine Image
      - sky launch with Bare Metal (BM) A10 GPU instance
      - sky launch with standard imagess [ubuntu 20/22]
      - sky launch with Marketplace images [ubuntu 20/22]
      - sky stop
      - sky start
      - sky down
      - sky down already stopped cluster
      - sky status
      - sky status --refresh
      - sky autostop 
- [ ] All smoke tests: `pytest tests/test_smoke.py` 
      - N/A
- [ ] Relevant individual smoke tests: `pytest tests/test_smoke.py::test_fill_in_the_name` 
      - N/A
- [ ] Backward compatibility tests: `bash tests/backward_comaptibility_tests.sh`
      - Passed


HysunHe: _"1. Have the STOPPING cluster here, and handle the STOPPING status in the create_node when reusing the existing cluster. Here is how we do it for AWS
2. In terminate_node we wait the instance to be STOPPED before we return. Here is how we do it for GCP"_

I used the second approach to stop instances before but commented out afterwards because it will cause stop process slow. Considering it is a small chance for reusing a "STOPPING" instance, the pt.1 option looks better. So this time I add this logic to the create_node function. 

HysunHe: Hope this time the format check on the sever can be passed:

_(sky) hysunhe@HYHE-PF1ZGYCQ:~/projects/skypilot/skypilot_oci$ ./format.sh --all
SkyPilot yapf: Done
SkyPilot Black:
reformatted sky/skylet/providers/oci/utils.py
reformatted sky/skylet/providers/oci/config.py
reformatted sky/skylet/providers/oci/query_helper.py
reformatted sky/skylet/providers/oci/node_provider.py

All done! ✨ 🍰 ✨
4 files reformatted, 19 files left unchanged.
SkyPilot mypy:
Success: no issues found in 109 source files
Sky Pylint:

------------------------------------
Your code has been rated at 10.00/10

(sky) hysunhe@HYHE-PF1ZGYCQ:~/projects/skypilot/skypilot_oci$ SKYPILOT_DISABLE_USAGE_COLLECTION=1 SKYPILOT_SKIP_CLOUD_IDENTITY_CHECK=1 pytest tests/test_cli.py
bringing up nodes...
....
4 passed, 16 warnings in 20.50s_
HysunHe: I'm going to add some instructions.
Michaelvll: Continue from the previous comment:
I fixed the name for the API key file and added the `~/.sky/config.yaml` with the `us-sanjose-1` vcn_id setup
```
oci:
  default:
    compartment_ocid: ocid1.compartment.oc1..random-hash
  us-sanjose-1:
    vcn_subnet: ocid1.subnet.oc1.us-sanjose-1.random-hash
```
7. When I `sky launch` again, it still tries to launch in `ap-seoul-1` and the following error occurs:
```
...
  File "/skypilot_oci/sky/skylet/providers/oci/node_provider.py", line 88, in _get_filtered_nodes
    insts = oci_query_helper.query_instances_by_tags(
  File "/skypilot_oci/sky/skylet/providers/oci/utils.py", line 15, in wrapper
    return f(*args, **kwargs)
  File "/skypilot_oci/sky/skylet/providers/oci/query_helper.py", line 47, in query_instances_by_tags
    region).search_resources(qv)
  File "/opt/conda/lib/python3.10/site-packages/oci/resource_search/resource_search_client.py", line 398, in search_resources
    return retry_strategy.make_retrying_call(
  File "/opt/conda/lib/python3.10/site-packages/oci/retry/retry.py", line 308, in make_retrying_call
    response = func_ref(*func_args, **func_kwargs)
  File "/opt/conda/lib/python3.10/site-packages/oci/base_client.py", line 531, in call_api
    response = self.request(request, allow_control_chars, operation_name, api_reference_link)
  File "/opt/conda/lib/python3.10/site-packages/circuitbreaker.py", line 159, in wrapper
    return call(function, *args, **kwargs)
  File "/opt/conda/lib/python3.10/site-packages/circuitbreaker.py", line 170, in call
    return func(*args, **kwargs)
  File "/opt/conda/lib/python3.10/site-packages/oci/base_client.py", line 718, in request
    self.raise_service_error(request, response, service_code, message, operation_name, api_reference_link, target_service, request_endpoint, client_version, timestamp, deserialized_data)
  File "/opt/conda/lib/python3.10/site-packages/oci/base_client.py", line 875, in raise_service_error
    raise exceptions.ServiceError(
oci.exceptions.ServiceError: {'target_service': 'resource_search', 'status': 401, 'code': 'NotAuthenticated', 'opc-request-id': '622B4803C4B447DC86F541D7CECBF635/E81F55FE6999B4DA9D5F0276096BF159/C941A3AFBCBC9027EC4ACA0CE05EB054', 'message': 'The required
 information to complete authentication was not provided or was incorrect.', 'operation_name': 'search_resources', 'timestamp': '2023-05-12T08:21:57.367795+00:00', 'client_version': 'Oracle-PythonSDK/2.101.0', 'request_endpoint': 'POST https://query.ap-s
eoul-1.oci.oraclecloud.com/20180409/resources', 'logging_tips': 'To get more info on the failing request, refer to https://docs.oracle.com/en-us/iaas/tools/python/latest/logging.html for ways to log the request/response details.', 'troubleshooting_tips':
 "See https://docs.oracle.com/iaas/Content/API/References/apierrors.htm#apierrors_401__401_notauthenticated for more information about resolving this error. Also see https://docs.oracle.com/iaas/api/#/en/search/20180409/ResourceSummary/SearchResources fo
r details on this operation's requirements. If you are unable to resolve this resource_search issue, please contact Oracle support and provide them this full error message."}
```


HysunHe: > 

Now the .sky/config.yaml is optional to OCI. Not sure if VCN configuration has issue, so please move .sky/config.yaml and try again.
HysunHe: > I am trying out this PR to actually launch clusters on OCI.
> 
> Several problems (some of them are UX issues):
> 
> 1. `sky check` does not show the correct output, when the `~/.oci/config` does not exist
> 
> <img alt="image" width="1726" src="https://user-images.githubusercontent.com/6753189/237904591-21640768-7697-48cf-8e1a-bdb58d21a941.png"> 2. `sky check` output does not align with other cloud (with some unnecessary outputs, such as: `Checking oci...`) <img alt="image" width="698" src="https://user-images.githubusercontent.com/6753189/237905419-9cb9de16-2e95-4b5d-b765-4f5e730f67ab.png">
> 3. The default CPU instance does not match the other clouds' setting (8-core CPU). We should at least match the CPU cores. (see the comments below)
> 4. `sky launch -c min --cpus 8 echo hi` the optimizer output is lower case instead of upper case.
> 
> <img alt="image" width="1009" src="https://user-images.githubusercontent.com/6753189/237907979-23d25769-7c64-46b0-87b0-4ebbd1caf140.png">
> 5. When type `yes` for the `sky launch`, the following error occurs, although I have specified the path to the API key in the `~/.oci/config` as suggested by the instruction of OCI cloud console.
> 
> ```
> cp: cannot stat '/root/.oci/oci_api_key.pem': No such file or directory
> ```
> 
> my `~/.oci/config` file:
> 
> ```
> [DEFAULT]
> user=ocid1.user.oc1..random-hash
> fingerprint=some-random-fingerprint
> tenancy=ocid1.tenancy.oc1..random-hash
> region=us-sanjose-1
> key_file=~/.oci/oracleidentitycloudservice_username-date.pem
> ```
> 
> 6. Another error also showed up in the output, which I believe to be the problem that we require the `~/.sky/config.yaml` for user account-specific configs. I am not in favor of this, as all the other clouds do not need that.
> 
> ```
> cp: cannot stat '/root/.sky/config.yaml': No such file or directory
> ```
> 
> I will manually change those path name, add those missing files, and try it again later, but would like to send out the issues first for faster iteration.
> 
> Based on our previous observation, the initial user experience is very important to get new users for the cloud. I would recommend fixing the above issues in this PR.

Those hardcoded path (.j2 ray template file) are removed. This kind of issue will never happen now.
HysunHe: Please kindly help merge the catalog PR first. otherwise the list_accelerators case cannot be passed.
Michaelvll: Thanks for the awesome work supporting OCI in SkyPilot @HysunHe! The latest code looks good to me. Merging. : )

Patch:
diff --git a/examples/oci/config.yaml b/examples/oci/config.yaml
new file mode 100644
index 000000000..498db946d
--- /dev/null
+++ b/examples/oci/config.yaml
@@ -0,0 +1,11 @@
+oci:
+  default:
+    compartment_ocid: ocid1.compartment.oc1..aaaaaaaahr7aicqtodxmcfor6pbqn3hvsngpftozyxzqw36gj4kh3w3kkj4q
+    #image_tag_general: skypilot:oci-ubuntu-20_04
+    #image_tag_gpu: skypilot:oci-ubuntu-NVIDIA-VMI-20_04
+
+  ap-seoul-1:
+    vcn_subnet: ocid1.subnet.oc1.ap-seoul-1.aaaaaaaa5c6wndifsij6yfyfehmi3tazn6mvhhiewqmajzcrlryurnl7nuja
+
+  us-ashburn-1:
+    vcn_subnet: ocid1.subnet.oc1.iad.aaaaaaaafbj7i3aqc4ofjaapa5edakde6g4ea2yaslcsay32cthp7qo55pxa
diff --git a/examples/oci/oci_cpu-sky-preemptible.yaml b/examples/oci/oci_cpu-sky-preemptible.yaml
new file mode 100644
index 000000000..e3bfc38fe
--- /dev/null
+++ b/examples/oci/oci_cpu-sky-preemptible.yaml
@@ -0,0 +1,39 @@
+name: cpu-task
+
+resources:
+  # Optional; if left out, automatically pick the cheapest cloud.
+  cloud: oci
+
+  region: ap-seoul-1
+  
+  zone: bxtG:AP-SEOUL-1-AD-1
+  
+  #instance_type: VM.Standard.E4.Flex
+
+  cpus: 2
+
+  image_id: skypilot:oci-ubuntu-20_04
+
+  use_spot: True
+
+# Working directory (optional) containing the project codebase.
+# Its contents are synced to ~/sky_workdir/ on the cluster.
+workdir: .
+
+num_nodes: 2
+
+# Typical use: pip install -r requirements.txt
+# Invoked under the workdir (i.e., can use its files).
+setup: |
+  echo "*** Running setup for the task. ***"
+
+# Typical use: make use of resources, such as running training.
+# Invoked under the workdir (i.e., can use its files).
+run: |
+  echo "*** Running the task on OCI (Preemptible) ***"
+  timestamp=$(date +%s)
+  for i in {1..10}; do
+    echo "$timestamp $i"
+    sleep 1
+  done
+  echo "The task is completed."
diff --git a/examples/oci/oci_cpu-sky.yaml b/examples/oci/oci_cpu-sky.yaml
new file mode 100644
index 000000000..482d4e2a8
--- /dev/null
+++ b/examples/oci/oci_cpu-sky.yaml
@@ -0,0 +1,39 @@
+name: cpu-task
+
+resources:
+  # Optional; if left out, automatically pick the cheapest cloud.
+  cloud: oci
+
+  region: ap-seoul-1
+  
+  zone: bxtG:AP-SEOUL-1-AD-1
+  
+  instance_type: VM.Standard.E4.Flex
+
+  cpus: 2
+
+  image_id: skypilot:oci-ubuntu-20_04
+
+  #use_spot: True
+
+# Working directory (optional) containing the project codebase.
+# Its contents are synced to ~/sky_workdir/ on the cluster.
+workdir: .
+
+num_nodes: 2
+
+# Typical use: pip install -r requirements.txt
+# Invoked under the workdir (i.e., can use its files).
+setup: |
+  echo "*** Running setup for the task. ***"
+
+# Typical use: make use of resources, such as running training.
+# Invoked under the workdir (i.e., can use its files).
+run: |
+  echo "*** Running the task on OCI ***"
+  timestamp=$(date +%s)
+  for i in {1..10}; do
+    echo "$timestamp $i"
+    sleep 1
+  done
+  echo "The task is completed."
diff --git a/examples/oci/oci_gpu-sky.yaml b/examples/oci/oci_gpu-sky.yaml
new file mode 100644
index 000000000..5d0c31721
--- /dev/null
+++ b/examples/oci/oci_gpu-sky.yaml
@@ -0,0 +1,29 @@
+name: cpu-task
+
+resources:
+  # Optional; if left out, automatically pick the cheapest cloud.
+  cloud: oci
+
+  region: ap-seoul-1
+  
+  zone: bxtG:AP-SEOUL-1-AD-1
+  
+  instance_type: VM.GPU.A10.1
+
+  image_id: skypilot:oci-ubuntu-NVIDIA-VMI-20_04
+
+# Working directory (optional) containing the project codebase.
+# Its contents are synced to ~/sky_workdir/ on the cluster.
+workdir: .
+
+num_nodes: 1
+
+# Typical use: pip install -r requirements.txt
+# Invoked under the workdir (i.e., can use its files).
+setup: |
+  echo "*** Running setup. ***"
+
+# Typical use: make use of resources, such as running training.
+# Invoked under the workdir (i.e., can use its files).
+run: |
+  echo "*** Hello, SkyPilot for OCI ***"
diff --git a/format.sh b/format.sh
index 8857fe759..9100aeadc 100755
--- a/format.sh
+++ b/format.sh
@@ -47,6 +47,7 @@ YAPF_FLAGS=(
 )
 
 YAPF_EXCLUDES=(
+    '--exclude' 'build/**'
     '--exclude' 'sky/skylet/providers/aws/**'
     '--exclude' 'sky/skylet/providers/gcp/**'
     '--exclude' 'sky/skylet/providers/azure/**'
diff --git a/sky/__init__.py b/sky/__init__.py
index e0559eab8..036fe71b7 100644
--- a/sky/__init__.py
+++ b/sky/__init__.py
@@ -32,6 +32,7 @@ GCP = clouds.GCP
 Lambda = clouds.Lambda
 SCP = clouds.SCP
 Local = clouds.Local
+OCI = clouds.OCI
 optimize = Optimizer.optimize
 
 __all__ = [
@@ -43,6 +44,7 @@ __all__ = [
     'Lambda',
     'SCP',
     'Local',
+    'OCI',
     'Optimizer',
     'OptimizeTarget',
     'backends',
diff --git a/sky/adaptors/oci.py b/sky/adaptors/oci.py
new file mode 100644
index 000000000..abd9dc353
--- /dev/null
+++ b/sky/adaptors/oci.py
@@ -0,0 +1,79 @@
+"""Oracle OCI cloud adaptor"""
+
+import functools
+import os
+
+CONFIG_PATH = '~/.oci/config'
+ENV_VAR_OCI_CONFIG = 'OCI_CONFIG'
+
+oci = None
+
+
+def import_package(func):
+
+    @functools.wraps(func)
+    def wrapper(*args, **kwargs):
+        global oci
+        if oci is None:
+            try:
+                # pylint: disable=import-outside-toplevel
+                import oci as _oci
+                oci = _oci
+            except ImportError:
+                raise ImportError('Fail to import dependencies for OCI.'
+                                  'Try pip install "skypilot[oci]"') from None
+        return func(*args, **kwargs)
+
+    return wrapper
+
+
+def get_config_file() -> str:
+    conf_file_path = CONFIG_PATH
+    config_path_via_env_var = os.environ.get(ENV_VAR_OCI_CONFIG)
+    if config_path_via_env_var is not None:
+        conf_file_path = config_path_via_env_var
+    return conf_file_path
+
+
+@import_package
+def get_oci():
+    return oci
+
+
+@import_package
+def get_oci_config(region=None, profile='DEFAULT'):
+    conf_file_path = get_config_file()
+    oci_config = oci.config.from_file(file_location=conf_file_path,
+                                      profile_name=profile)
+    if region is not None:
+        oci_config['region'] = region
+    return oci_config
+
+
+@import_package
+def get_core_client(region=None, profile='DEFAULT'):
+    return oci.core.ComputeClient(get_oci_config(region, profile))
+
+
+@import_package
+def get_net_client(region=None, profile='DEFAULT'):
+    return oci.core.VirtualNetworkClient(get_oci_config(region, profile))
+
+
+@import_package
+def get_search_client(region=None, profile='DEFAULT'):
+    return oci.resource_search.ResourceSearchClient(
+        get_oci_config(region, profile))
+
+
+@import_package
+def get_identity_client(region=None, profile='DEFAULT'):
+    return oci.identity.IdentityClient(get_oci_config(region, profile))
+
+
+@import_package
+def service_exception():
+    """OCI service exception."""
+    # pylint: disable=import-outside-toplevel
+    from oci.exceptions import ServiceError
+    return ServiceError
diff --git a/sky/authentication.py b/sky/authentication.py
index d43cf41db..4a6d96a9f 100644
--- a/sky/authentication.py
+++ b/sky/authentication.py
@@ -391,6 +391,27 @@ def setup_ibm_authentication(config):
     return config
 
 
+# Apr, 2023 by Hysun(hysun.he@oracle.com): Added support for OCI
+def setup_oci_authentication(config: Dict[str, Any]) -> Dict[str, Any]:
+    _, public_key_path = get_or_generate_keys()
+    with open(public_key_path, 'r') as f:
+        public_key = f.read()
+
+    # Need to use ~ relative path because Ray uses the same
+    # path for finding the public key path on both local and head node.
+    config['auth']['ssh_public_key'] = PUBLIC_SSH_KEY_PATH
+
+    file_mounts = config['file_mounts']
+    file_mounts[PUBLIC_SSH_KEY_PATH] = PUBLIC_SSH_KEY_PATH
+    config['file_mounts'] = file_mounts
+
+    for node_type in config['available_node_types']:
+        config['available_node_types'][node_type]['node_config'][
+            'AuthorizedKey'] = public_key
+
+    return config
+
+
 def setup_scp_authentication(config: Dict[str, Any]) -> Dict[str, Any]:
     private_key_path, public_key_path = get_or_generate_keys()
     config['auth']['ssh_private_key'] = private_key_path
diff --git a/sky/backends/backend_utils.py b/sky/backends/backend_utils.py
index ef264945f..2ba88bc05 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -1035,6 +1035,8 @@ def _add_auth_to_cluster_config(cloud: clouds.Cloud, cluster_config_file: str):
         config = auth.setup_ibm_authentication(config)
     elif isinstance(cloud, clouds.SCP):
         config = auth.setup_scp_authentication(config)
+    elif isinstance(cloud, clouds.OCI):
+        config = auth.setup_oci_authentication(config)
     else:
         assert isinstance(cloud, clouds.Local), cloud
         # Local cluster case, authentication is already filled by the user
@@ -1859,6 +1861,43 @@ def _query_status_scp(
     return status_list
 
 
+#Apr, 2023 by Hysun(hysun.he@oracle.com): Added support for OCI
+def _query_status_oci(
+        cluster: str,
+        ray_config: Dict[str, Any],  # pylint: disable=unused-argument
+) -> List[global_user_state.ClusterStatus]:
+    region = ray_config['provider']['region']
+
+    # Check the lifecycleState definition from the page
+    # https://docs.oracle.com/en-us/iaas/api/#/en/iaas/latest/Instance/
+    status_map = {
+        'PROVISIONING': global_user_state.ClusterStatus.INIT,
+        'STARTING': global_user_state.ClusterStatus.INIT,
+        'RUNNING': global_user_state.ClusterStatus.UP,
+        'STOPPING': global_user_state.ClusterStatus.STOPPED,
+        'STOPPED': global_user_state.ClusterStatus.STOPPED,
+        'TERMINATED': None,
+        'TERMINATING': None,
+    }
+
+    # pylint: disable=import-outside-toplevel
+    from sky.skylet.providers.oci.query_helper import oci_query_helper
+    from ray.autoscaler.tags import (
+        TAG_RAY_CLUSTER_NAME,)
+
+    status_list = []
+    vms = oci_query_helper.query_instances_by_tags(
+        tag_filters={TAG_RAY_CLUSTER_NAME: cluster}, region=region)
+    for node in vms:
+        vm_status = node.lifecycle_state
+        if vm_status in status_map:
+            sky_status = status_map[vm_status]
+            if sky_status is not None:
+                status_list.append(sky_status)
+
+    return status_list
+
+
 _QUERY_STATUS_FUNCS = {
     'AWS': _query_status_aws,
     'GCP': _query_status_gcp,
@@ -1866,6 +1905,7 @@ _QUERY_STATUS_FUNCS = {
     'Lambda': _query_status_lambda,
     'IBM': _query_status_ibm,
     'SCP': _query_status_scp,
+    'OCI': _query_status_oci,
 }
 
 
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index 6141f74ae..a274c9d9b 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -73,6 +73,7 @@ _NODES_LAUNCHING_PROGRESS_TIMEOUT = {
     clouds.Lambda: 150,
     clouds.IBM: 160,
     clouds.Local: 90,
+    clouds.OCI: 300,
 }
 
 # Time gap between retries after failing to provision in all possible places.
@@ -128,6 +129,7 @@ def _get_cluster_config_template(cloud):
         clouds.IBM: 'ibm-ray.yml.j2',
         clouds.Local: 'local-ray.yml.j2',
         clouds.SCP: 'scp-ray.yml.j2',
+        clouds.OCI: 'oci-ray.yml.j2',
     }
     return cloud_to_template[type(cloud)]
 
@@ -956,6 +958,44 @@ class RetryingVmProvisioner(object):
         self._blocked_resources.add(
             launchable_resources.copy(region=region.name, zone=None))
 
+    # Apr, 2023 by Hysun(hysun.he@oracle.com): Added support for OCI
+    def _update_blocklist_on_oci_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 ('VcnSubnetNotFound' in s.strip()) or
+            ('oci.exceptions.ServiceError' in s.strip() and
+             ('NotAuthorizedOrNotFound' in s.strip() or 'CannotParseRequest' in
+              s.strip() or 'InternalError' in s.strip() or
+              'LimitExceeded' in s.strip() or 'NotAuthenticated' 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) in {region.name}:')
+        messages = '\n\t'.join(errors)
+        logger.warning(f'{style.DIM}\t{messages}{style.RESET_ALL}')
+
+        if zones is not None:
+            for zone in zones:
+                self._blocked_resources.add(
+                    launchable_resources.copy(zone=zone.name))
+
     def _update_blocklist_on_error(
             self, launchable_resources: 'resources_lib.Resources',
             region: 'clouds.Region', zones: Optional[List['clouds.Zone']],
@@ -993,6 +1033,7 @@ class RetryingVmProvisioner(object):
             clouds.IBM: self._update_blocklist_on_ibm_error,
             clouds.SCP: self._update_blocklist_on_scp_error,
             clouds.Local: self._update_blocklist_on_local_error,
+            clouds.OCI: self._update_blocklist_on_oci_error,
         }
         cloud = launchable_resources.cloud
         cloud_type = type(cloud)
@@ -3247,6 +3288,24 @@ class CloudVmRayBackend(backends.Backend['CloudVmRayResourceHandle']):
                 stdout = ''
                 stderr = str(e)
 
+        # Apr, 2023 by Hysun(hysun.he@oracle.com): Added support for OCI
+        # May, 2023 by Hysun: Allow terminate INIT cluster which may have
+        # some instances provisioning in backgroud but not completed.
+        elif (isinstance(cloud, clouds.OCI) and terminate and
+              prev_cluster_status in (global_user_state.ClusterStatus.STOPPED,
+                                      global_user_state.ClusterStatus.INIT)):
+            region = config['provider']['region']
+
+            # pylint: disable=import-outside-toplevel
+            from sky.skylet.providers.oci.query_helper import oci_query_helper
+            from ray.autoscaler.tags import TAG_RAY_CLUSTER_NAME
+
+            # 0: All terminated successfully, failed count otherwise
+            returncode = oci_query_helper.terminate_instances_by_tags(
+                {TAG_RAY_CLUSTER_NAME: cluster_name}, region)
+
+            # To avoid undefined local varaiables error.
+            stdout = stderr = ''
         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 9b1f7868d..c3e3fea8d 100644
--- a/sky/clouds/__init__.py
+++ b/sky/clouds/__init__.py
@@ -11,6 +11,7 @@ from sky.clouds.lambda_cloud import Lambda
 from sky.clouds.local import Local
 from sky.clouds.ibm import IBM
 from sky.clouds.scp import SCP
+from sky.clouds.oci import OCI
 
 __all__ = [
     'IBM',
@@ -21,6 +22,7 @@ __all__ = [
     'Lambda',
     'Local',
     'SCP',
+    'OCI',
     'CloudImplementationFeatures',
     'Region',
     'Zone',
diff --git a/sky/clouds/oci.py b/sky/clouds/oci.py
new file mode 100644
index 000000000..81682c2e8
--- /dev/null
+++ b/sky/clouds/oci.py
@@ -0,0 +1,436 @@
+"""
+Oracle Cloud Infrastructure (OCI)
+
+History:
+ - Hysun He (hysun.he@oracle.com) @ Apr, 2023: Initial implementation
+ - Hysun He (hysun.he@oracle.com) @ May 4, 2023: Support use the default
+   image_id (configurable) if no image_id specified in the task yaml.
+"""
+import os
+import json
+import typing
+import logging
+from typing import Dict, Iterator, List, Optional, Tuple
+
+from sky import clouds
+from sky.clouds import service_catalog
+from sky import exceptions
+from sky.adaptors import oci as oci_adaptor
+from sky.skylet.providers.oci.config import oci_conf
+
+if typing.TYPE_CHECKING:
+    # Renaming to avoid shadowing variables.
+    from sky import resources as resources_lib
+
+logger = logging.getLogger(__name__)
+
+
+@clouds.CLOUD_REGISTRY.register
+class OCI(clouds.Cloud):
+    """ OCI: Oracle Cloud Infrastructure """
+
+    _REPR = 'OCI'
+
+    _MAX_CLUSTER_NAME_LEN_LIMIT = 200
+
+    _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
+
+        regions = service_catalog.get_region_zones_for_instance_type(
+            instance_type, use_spot, 'oci')
+
+        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 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='oci')
+
+    @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[List[clouds.Zone]]:
+        del num_nodes  # unused
+        regions = cls.regions_with_offering(instance_type,
+                                            accelerators,
+                                            use_spot,
+                                            region=region,
+                                            zone=None)
+        for r in regions:
+            assert r.zones is not None, r
+            for zone in r.zones:
+                yield [zone]
+
+    def instance_type_to_hourly_cost(self,
+                                     instance_type: str,
+                                     use_spot: bool,
+                                     region: Optional[str] = None,
+                                     zone: Optional[str] = None) -> float:
+        return service_catalog.get_hourly_cost(instance_type,
+                                               use_spot=use_spot,
+                                               region=region,
+                                               zone=zone,
+                                               clouds='oci')
+
+    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
+        return 0.0
+
+    def get_egress_cost(self, num_gigabytes: float) -> float:
+        """
+        https://www.oracle.com/cis/cloud/networking/pricing/
+        """
+        # Free for first 10T (per month)
+        if num_gigabytes <= 10 * 1024:
+            return 0.0
+
+        # We need to calculate the egress cost by region.
+        # Fortunately, most of time, this cost looks not a
+        # big deal comparing to the price of GPU instances.
+        # # Calculate cost for over 10T (per month)
+        # logger.debug(f"* get_egress_cost. region {region}")
+        # if region in REGIONS_NorthAmerica_Europe_UK:
+        #     return (num_gigabytes - 10 * 1024) * 0.0085
+        # elif region in REGIONS_JAPAC_SouthAmerica:
+        #     return (num_gigabytes - 10 * 1024) * 0.025
+        # elif region in REGIONS_MidEast_Africa:
+        #     return (num_gigabytes - 10 * 1024) * 0.05
+        # else:
+        #     logger.debug(f"* ! Region {region} is not listed for cost calc!")
+        #     return 0.0
+        return (num_gigabytes - 10 * 1024) * 0.0085
+
+    def is_same_cloud(self, other: clouds.Cloud) -> bool:
+        # Returns true if the two clouds are the same cloud type.
+        return isinstance(other, OCI)
+
+    @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='oci')
+
+    @classmethod
+    def get_accelerators_from_instance_type(
+        cls,
+        instance_type: str,
+    ) -> Optional[Dict[str, int]]:
+        return service_catalog.get_accelerators_from_instance_type(
+            instance_type, clouds='oci')
+
+    @classmethod
+    def get_zone_shell_cmd(cls) -> Optional[str]:
+        return None
+
+    def make_deploy_resources_variables(
+            self, resources: 'resources_lib.Resources',
+            region: Optional['clouds.Region'],
+            zones: Optional[List['clouds.Zone']]) -> Dict[str, Optional[str]]:
+        assert region is not None, resources
+
+        acc_dict = self.get_accelerators_from_instance_type(
+            resources.instance_type)
+        if acc_dict is not None:
+            custom_resources = json.dumps(acc_dict, separators=(',', ':'))
+        else:
+            custom_resources = None
+
+        image_str = self._get_image_id(resources.image_id, region.name,
+                                       resources.instance_type)
+        image_cols = image_str.split(oci_conf.IMAGE_TAG_SPERATOR)
+        if len(image_cols) == 3:
+            image_id = image_cols[0]
+            listing_id = image_cols[1]
+            res_ver = image_cols[2]
+        else:
+            image_id = resources.image_id
+            listing_id = None
+            res_ver = None
+
+        cpus = resources.cpus
+        instance_type_arr = resources.instance_type.split(
+            oci_conf.INSTANCE_TYPE_RES_SPERATOR)
+        instance_type = instance_type_arr[0]
+
+        # Improvement:
+        # Fault-tolerant to the catalog file: special shapes does
+        # not need cpu/memory configuration, so ignore these info
+        # from the catalog file to avoid inconsistence (mainly due
+        # to the shape changed in future.)
+        if len(instance_type_arr) < 2:
+            cpus = None
+        else:
+            if cpus is None:
+                cpus, mems = OCI.get_vcpus_mem_from_instance_type(
+                    resources.instance_type)
+                resources = resources.copy(
+                    cpus=cpus,
+                    memory=mems,
+                )
+            if cpus is None and resources.instance_type.startswith(
+                    oci_conf.VM_PREFIX):
+                cpus = f'{oci_conf.DEFAULT_NUM_VCPUS}'
+
+        zone = resources.zone
+        if zone is None:
+            # If zone is not specified, try to get the first zone.
+            if zones is None:
+                regions = service_catalog.get_region_zones_for_instance_type(
+                    instance_type=resources.instance_type,
+                    use_spot=resources.use_spot,
+                    clouds='oci')
+                zones = [r for r in iter(regions) if r.name == region.name
+                        ][0].zones
+
+            if zones is not None:
+                zone = zones[0].name
+
+        return {
+            'instance_type': instance_type,
+            'custom_resources': custom_resources,
+            'region': region.name,
+            'cpus': cpus,
+            'memory': resources.memory,
+            'zone': zone,
+            'image': image_id,
+            'app_catalog_listing_id': listing_id,
+            'resource_version': res_ver,
+            'use_spot': resources.use_spot
+        }
+
+    def get_feasible_launchable_resources(self,
+                                          resources: 'resources_lib.Resources'):
+        if resources.instance_type is not None:
+            assert resources.is_launchable(), resources
+            resources = resources.copy(accelerators=None)
+            return ([resources], [])
+
+        def _make(instance_list):
+            resource_list = []
+            for instance_type in instance_list:
+                r = resources.copy(
+                    cloud=OCI(),
+                    instance_type=instance_type,
+                    # Setting this to None as OCI 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 = OCI.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,
+            use_spot=resources.use_spot,
+            cpus=resources.cpus,
+            memory=resources.memory,
+            region=resources.region,
+            zone=resources.zone,
+            clouds='oci')
+        if instance_list is None:
+            return ([], fuzzy_candidate_list)
+
+        return (_make(instance_list), fuzzy_candidate_list)
+
+    @classmethod
+    def check_credentials(cls) -> Tuple[bool, Optional[str]]:
+        """Checks if the user has access credentials to this cloud."""
+
+        try:
+            # pylint: disable=import-outside-toplevel,unused-import
+            import oci
+        except ImportError:
+            return False, ('oci is not installed. Install it with: '
+                           'pip install oci')
+
+        conf_file = oci_adaptor.get_config_file()
+
+        help_str = (f'Missing credential file at {conf_file}. '
+                    'To configure credentials, go to:\n'
+                    '      https://docs.oracle.com/en-us/iaas/Content/API/'
+                    'Concepts/apisigningkey.htm')
+        if not os.path.isfile(os.path.expanduser(conf_file)):
+            return (False, help_str)
+
+        try:
+            user = oci_adaptor.get_identity_client(
+                region=None, profile=oci_conf.get_profile()).get_user(
+                    oci_adaptor.get_oci_config()['user']).data
+            del user
+            # TODO[Hysun]: More privilege check can be added
+            return True, None
+        except oci_adaptor.service_exception():
+            return False, (f'OCI credential is not correctly set. '
+                           f'Check the credential file at {conf_file}')
+
+    def get_credential_file_mounts(self) -> Dict[str, str]:
+        """Returns a dict of credential file paths to mount paths."""
+        oci_cfg_file = oci_adaptor.get_config_file()
+        oci_cfg = oci_adaptor.get_oci_config()
+        api_key_file = oci_cfg[
+            'key_file'] if 'key_file' in oci_cfg else 'BadConf'
+        sky_cfg_file = oci_conf.get_sky_user_config_file()
+
+        # OCI config and API key file are mandatory
+        credential_files = [oci_cfg_file, api_key_file]
+
+        # Sky config file is optional
+        if os.path.exists(sky_cfg_file):
+            credential_files.append(sky_cfg_file)
+
+        file_mounts = {
+            f'{filename}': f'{filename}' for filename in credential_files
+        }
+
+        logger.debug(f'OCI credential file mounts: {file_mounts}')
+        return file_mounts
+
+    @classmethod
+    def get_current_user_identity(cls) -> Optional[List[str]]:
+        # NOTE: used for very advanced SkyPilot functionality
+        # Can implement later if desired
+        # If the user switches the compartment_ocid, the existing clusters
+        # might be leaked, as `sky status -r` will fail to find the original
+        # clusters in the new compartment and the identity check is missing.
+        return None
+
+    def instance_type_exists(self, instance_type: str) -> bool:
+        return service_catalog.instance_type_exists(instance_type, 'oci')
+
+    def validate_region_zone(self, region: Optional[str], zone: Optional[str]):
+        return service_catalog.validate_region_zone(region, zone, clouds='oci')
+
+    def accelerator_in_region_or_zone(self,
+                                      accelerator: str,
+                                      acc_count: int,
+                                      region: Optional[str] = None,
+                                      zone: Optional[str] = None) -> bool:
+        return service_catalog.accelerator_in_region_or_zone(
+            accelerator, acc_count, region, zone, 'oci')
+
+    def get_image_size(self, image_id: str, region: Optional[str]) -> float:
+        # We ignore checking the image size because most of situations the
+        # boot volume size is larger than the image size. For specific rare
+        # situations, the configuration/setup commands should make sure the
+        # correct size of the disk.
+        return 0
+
+    def _get_image_id(
+        self,
+        image_id: Optional[Dict[Optional[str], str]],
+        region_name: str,
+        instance_type: str,
+    ) -> str:
+        if image_id is None:
+            return self._get_default_image(region_name=region_name,
+                                           instance_type=instance_type)
+        if None in image_id:
+            image_id_str = image_id[None]
+        else:
+            assert region_name in image_id, image_id
+            image_id_str = image_id[region_name]
+        if image_id_str.startswith('skypilot:'):
+            image_id_str = service_catalog.get_image_id_from_tag(image_id_str,
+                                                                 region_name,
+                                                                 clouds='oci')
+            if image_id_str is None:
+                logger.critical(
+                    '! Real image_id not found! - {region_name}:{image_id}')
+                # Raise ResourcesUnavailableError to make sure the failover
+                # in CloudVMRayBackend will be correctly triggered.
+                # TODO(zhwu): This is a information leakage to the cloud
+                # implementor, we need to find a better way to handle this.
+                raise exceptions.ResourcesUnavailableError(
+                    '! ERR: No image found in catalog for region '
+                    f'{region_name}. Try setting a valid image_id.')
+
+        logger.debug(f'Got real image_id {image_id_str}')
+        return image_id_str
+
+    def _get_default_image(self, region_name: str, instance_type: str) -> str:
+        acc = self.get_accelerators_from_instance_type(instance_type)
+
+        if acc is None:
+            image_tag = oci_conf.get_default_image_tag()
+            image_id_str = service_catalog.get_image_id_from_tag(image_tag,
+                                                                 region_name,
+                                                                 clouds='oci')
+        else:
+            assert len(acc) == 1, acc
+            image_tag = oci_conf.get_default_gpu_image_tag()
+            image_id_str = service_catalog.get_image_id_from_tag(image_tag,
+                                                                 region_name,
+                                                                 clouds='oci')
+
+        if image_id_str is not None:
+            logger.debug(
+                f'Got default image_id {image_id_str} from tag {image_tag}')
+            return image_id_str
+
+        # Raise ResourcesUnavailableError to make sure the failover in
+        # CloudVMRayBackend will be correctly triggered.
+        # TODO(zhwu): This is a information leakage to the cloud implementor,
+        # we need to find a better way to handle this.
+        raise exceptions.ResourcesUnavailableError(
+            'ERR: No image found in catalog for region '
+            f'{region_name}. Try update your default image_id settings.')
diff --git a/sky/clouds/service_catalog/__init__.py b/sky/clouds/service_catalog/__init__.py
index 1d301fe3a..3116831b8 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', 'ibm', 'lambda', 'scp')
+_ALL_CLOUDS = ('aws', 'azure', 'gcp', 'ibm', 'lambda', 'scp', 'oci')
 
 
 def _map_clouds_catalog(clouds: CloudFilter, method_name: str, *args, **kwargs):
diff --git a/sky/clouds/service_catalog/oci_catalog.py b/sky/clouds/service_catalog/oci_catalog.py
new file mode 100644
index 000000000..e5ea00f4e
--- /dev/null
+++ b/sky/clouds/service_catalog/oci_catalog.py
@@ -0,0 +1,139 @@
+"""OCI Offerings Catalog.
+
+This module loads the service catalog file and can be used to query
+instance types and pricing information for OCI.
+
+History:
+ - Hysun He (hysun.he@oracle.com) @ Apr, 2023: Initial implementation
+"""
+
+import typing
+import logging
+from typing import Dict, List, Optional, Tuple
+from sky.clouds.service_catalog import common
+from sky.skylet.providers.oci.config import oci_conf
+
+if typing.TYPE_CHECKING:
+    from sky.clouds import cloud
+
+logger = logging.getLogger(__name__)
+
+_df = common.read_catalog('oci/vms.csv')
+_image_df = common.read_catalog('oci/images.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]) -> Tuple[Optional[str], Optional[str]]:
+    return common.validate_region_zone_impl('oci', _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:
+    """Returns the cost, or the cheapest cost among all zones for spot."""
+    return common.get_hourly_cost_impl(_df, instance_type, use_spot, region,
+                                       zone)
+
+
+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:
+        cpus = f'{oci_conf.DEFAULT_NUM_VCPUS}+'
+
+    if memory is None:
+        memory_gb_or_ratio = f'{oci_conf.DEFAULT_MEMORY_CPU_RATIO}x'
+    else:
+        memory_gb_or_ratio = memory
+
+    instance_type_prefix = tuple(
+        f'{family}' for family in oci_conf.DEFAULT_INSTANCE_FAMILY)
+
+    df = _df[_df['InstanceType'].notna()]
+    df = df[df['InstanceType'].str.startswith(instance_type_prefix)]
+
+    logger.debug(f'# get_default_instance_type: {df}')
+    return common.get_instance_type_for_cpus_mem_impl(df, cpus,
+                                                      memory_gb_or_ratio)
+
+
+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],
+        quantity_filter: Optional[int],
+        case_sensitive: bool = True
+) -> Dict[str, List[common.InstanceTypeInfo]]:
+    """Returns all instance types in OCI offering GPUs."""
+    return common.list_accelerators_impl('OCI', _df, gpus_only, name_filter,
+                                         region_filter, quantity_filter,
+                                         case_sensitive)
+
+
+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_image_id_from_tag(tag: str, region: Optional[str]) -> Optional[str]:
+    """Returns the image id from the tag."""
+    image_str = common.get_image_id_from_tag_impl(_image_df, tag, region)
+    df = _image_df[_image_df['Tag'].str.fullmatch(tag)]
+    app_catalog_listing_id = df['AppCatalogListingId'].iloc[0]
+    resource_version = df['ResourceVersion'].iloc[0]
+    return (f'{image_str}{oci_conf.IMAGE_TAG_SPERATOR}{app_catalog_listing_id}'
+            f'{oci_conf.IMAGE_TAG_SPERATOR}{resource_version}')
+
+
+def is_image_tag_valid(tag: str, region: Optional[str]) -> bool:
+    """Returns whether the image tag is valid."""
+    return common.is_image_tag_valid_impl(_image_df, tag, region)
diff --git a/sky/resources.py b/sky/resources.py
index a7e8cfc94..b4e178303 100644
--- a/sky/resources.py
+++ b/sky/resources.py
@@ -647,13 +647,15 @@ class Resources:
                 raise ValueError(
                     'Cloud must be specified when image_id is provided.')
 
+        # Apr, 2023 by Hysun(hysun.he@oracle.com): Added support for OCI
         if not self._cloud.is_same_cloud(
-                clouds.IBM()) and not self._cloud.is_same_cloud(
-                    clouds.AWS()) and not self._cloud.is_same_cloud(
-                        clouds.GCP()):
+                clouds.AWS()) and not self._cloud.is_same_cloud(
+                    clouds.GCP()) and not self._cloud.is_same_cloud(
+                        clouds.IBM()) and not self._cloud.is_same_cloud(
+                            clouds.OCI()):
             with ux_utils.print_exception_no_traceback():
                 raise ValueError(
-                    'image_id is only supported for AWS, GCP and IBM, please '
+                    'image_id is only supported for AWS/GCP/IBM/OCI, please '
                     'explicitly specify the cloud.')
 
         if self._region is not None:
@@ -899,7 +901,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/IBM.')
+                           'supports AWS/GCP/IBM/OCI.')
             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 347f3c558..8dc4c9235 100644
--- a/sky/setup_files/MANIFEST.in
+++ b/sky/setup_files/MANIFEST.in
@@ -7,6 +7,7 @@ include sky/skylet/providers/gcp/*
 include sky/skylet/providers/lambda_cloud/*
 include sky/skylet/providers/ibm/*
 include sky/skylet/providers/scp/*
+include sky/skylet/providers/oci/*
 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 d2919e5ea..3a7f409c3 100644
--- a/sky/setup_files/setup.py
+++ b/sky/setup_files/setup.py
@@ -136,6 +136,7 @@ extras_require: Dict[str, List[str]] = {
     'lambda': [],
     'cloudflare': aws_dependencies,
     'scp': [],
+    'oci': ['oci'],
 }
 
 extras_require['all'] = sum(extras_require.values(), [])
diff --git a/sky/skylet/providers/oci/__init__.py b/sky/skylet/providers/oci/__init__.py
new file mode 100644
index 000000000..f7c3aa255
--- /dev/null
+++ b/sky/skylet/providers/oci/__init__.py
@@ -0,0 +1,2 @@
+"""OCI node provider"""
+from sky.skylet.providers.oci.node_provider import OCINodeProvider
diff --git a/sky/skylet/providers/oci/config.py b/sky/skylet/providers/oci/config.py
new file mode 100644
index 000000000..f4b4d7ad8
--- /dev/null
+++ b/sky/skylet/providers/oci/config.py
@@ -0,0 +1,87 @@
+"""
+OCI Configuration.
+History:
+ - Hysun He (hysun.he@oracle.com) @ Apr, 2023: Initial implementation
+ 
+"""
+import logging
+import os
+from sky import skypilot_config
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+
+class oci_conf:
+    IMAGE_TAG_SPERATOR = "|"
+    INSTANCE_TYPE_RES_SPERATOR = "$_"
+    CPU_MEM_SPERATOR = "_"
+
+    DEFAULT_NUM_VCPUS = 8
+    DEFAULT_MEMORY_CPU_RATIO = 4
+
+    VM_PREFIX = "VM.Standard"
+    DEFAULT_INSTANCE_FAMILY = [
+        # CPU: AMD, Memory: 8 GiB RAM per 1 vCPU;
+        f"{VM_PREFIX}.E",
+        # CPU: Intel, Memory: 8 GiB RAM per 1 vCPU;
+        f"{VM_PREFIX}3",
+        # CPU: ARM, Memory: 6 GiB RAM per 1 vCPU;
+        # f'{VM_PREFIX}.A',
+    ]
+
+    COMPARTMENT = "skypilot_compartment"
+    VCN_NAME = "skypilot_vcn"
+    VCN_DNS_LABEL = "skypilotvcn"
+    VCN_INTERNET_GATEWAY_NAME = "skypilot_vcn_internet_gateway"
+    VCN_SUBNET_NAME = "skypilot_subnet"
+    VCN_CIDR_INTERNET = "0.0.0.0/0"
+    VCN_CIDR = "192.168.0.0/16"
+    VCN_SUBNET_CIDR = "192.168.0.0/18"
+
+    MAX_RETRY_COUNT = 3
+    RETRY_INTERVAL_BASE_SECONDS = 5
+
+    @classmethod
+    def get_compartment(cls, region):
+        # Allow task(cluster)-specific compartment/VCN parameters.
+        default_compartment_ocid = skypilot_config.get_nested(
+            ("oci", "default", "compartment_ocid"), None)
+        compartment = skypilot_config.get_nested(
+            ("oci", region, "compartment_ocid"), default_compartment_ocid)
+        return compartment
+
+    @classmethod
+    def get_vcn_subnet(cls, region):
+        vcn = skypilot_config.get_nested(("oci", region, "vcn_subnet"), None)
+        return vcn
+
+    @classmethod
+    def get_default_gpu_image_tag(cls) -> str:
+        # Get the default image tag (for gpu instances). Instead of hardcoding, we give a choice to set the
+        # default image tag (for gpu instances) in the sky's user-config file (if not specified, use the hardcode
+        # one at last)
+        return skypilot_config.get_nested(("oci", "default", "image_tag_gpu"),
+                                          "skypilot:gpu-ubuntu-2004")
+
+    @classmethod
+    def get_default_image_tag(cls) -> str:
+        # Get the default image tag. Instead of hardcoding, we give a choice to set the default image tag
+        # in the sky's user-config file. (if not specified, use the hardcode one at last)
+        return skypilot_config.get_nested(
+            ("oci", "default", "image_tag_general"), "skypilot:cpu-ubuntu-2004")
+
+    @classmethod
+    def get_sky_user_config_file(cls) -> str:
+        config_path_via_env_var = os.environ.get(
+            skypilot_config.ENV_VAR_SKYPILOT_CONFIG)
+        if config_path_via_env_var is not None:
+            config_path = config_path_via_env_var
+        else:
+            config_path = skypilot_config.CONFIG_PATH
+        return config_path
+
+    @classmethod
+    def get_profile(cls) -> str:
+        return skypilot_config.get_nested(
+            ("oci", "default", "oci_config_profile"), "DEFAULT")
diff --git a/sky/skylet/providers/oci/node_provider.py b/sky/skylet/providers/oci/node_provider.py
new file mode 100644
index 000000000..9ebb9e138
--- /dev/null
+++ b/sky/skylet/providers/oci/node_provider.py
@@ -0,0 +1,483 @@
+"""OCI Node Provider.
+
+Node provider is called by the Ray Autoscaler to provision new compute
+resources (head / worker nodes).
+
+To show debug messages, export SKYPILOT_DEBUG=1 
+
+History:
+ - Hysun He (hysun.he@oracle.com) @ Apr, 2023: Initial implementation
+ 
+"""
+
+import logging
+import time
+import threading
+import copy
+
+from datetime import datetime
+from sky.skylet.providers.oci.config import oci_conf
+from sky.skylet.providers.oci import utils
+from sky.skylet.providers.oci.query_helper import oci_query_helper
+from sky.adaptors import oci as oci_adaptor
+
+from ray.autoscaler.node_provider import NodeProvider
+from ray.autoscaler.tags import (
+    TAG_RAY_CLUSTER_NAME,
+    TAG_RAY_NODE_KIND,
+    TAG_RAY_USER_NODE_TYPE,
+    TAG_RAY_LAUNCH_CONFIG,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def synchronized(f):
+
+    def wrapper(self, *args, **kwargs):
+        self.lock.acquire()
+        try:
+            return f(self, *args, **kwargs)
+        finally:
+            self.lock.release()
+
+    return wrapper
+
+
+class OCINodeProvider(NodeProvider):
+    """Node Provider for OracleCloud (OCI)."""
+
+    def __init__(self, provider_config, cluster_name):
+        NodeProvider.__init__(self, provider_config, cluster_name)
+        self.lock = threading.RLock()
+        self.cached_nodes = {}
+        self.cache_stopped_nodes = provider_config.get("cache_stopped_nodes",
+                                                       True)
+        self.region = provider_config["region"]
+
+        # Do a read-ahead cache loading to improve performance.
+        self._get_filtered_nodes({})
+
+    @synchronized
+    def _get_filtered_nodes(self, tag_filters, force=False):
+        # Make sure the cluster_name is always an criterion
+        tag_filters = {**tag_filters, TAG_RAY_CLUSTER_NAME: self.cluster_name}
+
+        return_nodes = {}
+        if not force:
+            # Query cache first to reduce API call.
+            cache_hit = False
+            for k, node in self.cached_nodes.items():
+                tags = node["tags"]
+                unmatched_tags = [
+                    k for k, v in tag_filters.items()
+                    if k not in tags or v != tags[k]
+                ]
+                if len(unmatched_tags) == 0:
+                    return_nodes[k] = node
+                    cache_hit |= True
+
+            if cache_hit:
+                return return_nodes
+
+        insts = oci_query_helper.query_instances_by_tags(
+            tag_filters, self.region)
+        for inst in insts:
+            inst_id = inst.identifier
+            if inst_id in self.cached_nodes:
+                del self.cached_nodes[inst_id]
+
+            item = self.get_inst_obj({
+                "inst_id": inst_id,
+                "ad": inst.availability_domain,
+                "compartment": inst.compartment_id,
+                "lifecycle_state": inst.lifecycle_state,
+                "oci_tags": inst.freeform_tags,
+            })
+            return_nodes[inst_id] = item
+            self.cached_nodes[inst_id] = item
+
+        return return_nodes
+
+    @utils.debug_enabled(logger=logger)
+    def non_terminated_nodes(self, tag_filters):
+        """Return a list of node ids filtered by the specified tags dict.
+
+        This list must not include terminated nodes. For performance reasons,
+        providers are allowed to cache the result of a call to
+        non_terminated_nodes() to serve single-node queries
+        (e.g. is_running(node_id)). This means that non_terminated_nodes()
+        must be called again to refresh results.
+        """
+        VALIDITY_TAGS = [
+            TAG_RAY_CLUSTER_NAME,
+            TAG_RAY_NODE_KIND,
+            TAG_RAY_USER_NODE_TYPE,
+            TAG_RAY_LAUNCH_CONFIG,
+        ]
+        filters = {
+            tag: tag_filters[tag] for tag in VALIDITY_TAGS if tag in tag_filters
+        }
+
+        nodes = self._get_filtered_nodes(tag_filters=filters)
+        return [k for k, v in nodes.items() if v["status"] == "RUNNING"]
+
+    @utils.debug_enabled(logger=logger)
+    def is_running(self, node_id):
+        """Return whether the specified node is running."""
+        node = self._get_cached_node(node_id=node_id)
+        check_result = node is None or node["status"] == "RUNNING"
+
+        return check_result
+
+    @utils.debug_enabled(logger=logger)
+    def is_terminated(self, node_id):
+        """Return whether the specified node is terminated."""
+        node = self._get_cached_node(node_id=node_id)
+        check_result = ((node is None) or (node["status"] == "TERMINATED") or
+                        (node["status"] == "TERMINATING"))
+
+        return check_result
+
+    @utils.debug_enabled(logger=logger)
+    def node_tags(self, node_id):
+        return self.cached_nodes[node_id]["tags"]
+
+    @utils.debug_enabled(logger=logger)
+    def external_ip(self, node_id):
+        """Returns the external ip of the given node."""
+        return self._get_cached_node(node_id=node_id)["external_ip"]
+
+    @utils.debug_enabled(logger=logger)
+    def internal_ip(self, node_id):
+        """Returns the internal ip (Ray ip) of the given node."""
+        return self._get_cached_node(node_id=node_id)["internal_ip"]
+
+    @synchronized
+    @utils.debug_enabled(logger=logger)
+    def create_node(self, node_config, tags, count):
+        """Creates a number of nodes within the namespace."""
+        start_time = round(time.time() * 1000)
+        starting_insts = []
+        # Check first if it neccessary to create new nodes / start stopped nodes
+        VALIDITY_TAGS = [
+            TAG_RAY_CLUSTER_NAME,
+            TAG_RAY_NODE_KIND,
+            TAG_RAY_USER_NODE_TYPE,
+        ]
+        filters = {tag: tags[tag] for tag in VALIDITY_TAGS if tag in tags}
+        running_nodes = self.running_nodes(filters)
+        if len(running_nodes) > 0:
+            logger.info(
+                f"Running nodes found {len(running_nodes)}: {list(running_nodes)}. "
+                " Reuse existing running nodes. ")
+            count -= len(running_nodes)
+
+        if count <= 0:
+            logger.info(
+                f"No need to create new node since there are enough running nodes"
+            )
+            return
+
+        # Starting stopped nodes if cache_stopped_nodes=True
+        if self.cache_stopped_nodes:
+            logger.debug("Checking existing stopped nodes.")
+
+            filters_with_launch_config = copy.copy(filters)
+            if TAG_RAY_LAUNCH_CONFIG in tags:
+                filters_with_launch_config[TAG_RAY_LAUNCH_CONFIG] = tags[
+                    TAG_RAY_LAUNCH_CONFIG]
+
+            nodes_matching_launch_config = self.stopped_nodes(
+                filters_with_launch_config)
+            logger.debug(f"Found stopped nodes (with same launch config): "
+                         f"{len(nodes_matching_launch_config)}")
+
+            reuse_nodes = []
+            if len(nodes_matching_launch_config) >= count:
+                reuse_nodes = nodes_matching_launch_config[:count]
+            else:
+                nodes_all = self.stopped_nodes(filters)
+                logger.debug(f"Found stopped nodes (regardless launch config): "
+                             f"{len(nodes_all)}")
+                nodes_matching_launch_config_ids = [
+                    n["id"] for n in nodes_matching_launch_config
+                ]
+                nodes_non_matching_launch_config = [
+                    n for n in nodes_all
+                    if n["id"] not in nodes_matching_launch_config_ids
+                ]
+                reuse_nodes = (nodes_matching_launch_config +
+                               nodes_non_matching_launch_config)
+                reuse_nodes = reuse_nodes[:count]
+
+            logger.info(
+                f"Reusing nodes {len(reuse_nodes)}: {list(reuse_nodes)}. "
+                "To disable reuse, set `cache_stopped_nodes: False` "
+                "under `provider` in the cluster configuration.",)
+
+            for reuse_node in reuse_nodes:
+                if reuse_node["status"] == "STOPPING":
+                    get_instance_response = oci_adaptor.get_core_client(
+                        self.region, oci_conf.get_profile()).get_instance(
+                            instance_id=reuse_node["id"])
+                    oci_adaptor.get_oci().wait_until(
+                        oci_adaptor.get_core_client(self.region,
+                                                    oci_conf.get_profile()),
+                        get_instance_response,
+                        "lifecycle_state",
+                        "STOPPED",
+                    )
+
+            start_time1 = round(time.time() * 1000)
+            for matched_node in reuse_nodes:
+                matched_node_id = matched_node["id"]
+                instance_action_response = oci_adaptor.get_core_client(
+                    self.region, oci_conf.get_profile()).instance_action(
+                        instance_id=matched_node_id, action="START")
+
+                starting_inst = instance_action_response.data
+                starting_insts.append({
+                    "inst_id": starting_inst.id,
+                    "ad": starting_inst.availability_domain,
+                    "compartment": starting_inst.compartment_id,
+                    "lifecycle_state": starting_inst.lifecycle_state,
+                    "oci_tags": starting_inst.freeform_tags,
+                })
+            count -= len(reuse_nodes)
+
+            launch_stopped_time = round(time.time() * 1000) - start_time1
+            logger.debug(
+                "Time elapsed(Launch stopped): {0} milli-seconds.".format(
+                    launch_stopped_time))
+        # end if self.cache_stopped_nodes:...
+
+        # Let's create additional new nodes (if neccessary)
+        if count > 0:
+            compartment = oci_query_helper.find_compartment(self.region)
+            vcn = oci_query_helper.find_create_vcn_subnet(self.region)
+            if vcn is None:
+                raise RuntimeError("VcnSubnetNotFound Error!")
+
+            ocpu_count = 0
+            vcpu_str = node_config["VCPUs"]
+            instance_type_str = node_config["InstanceType"]
+
+            if vcpu_str is not None and vcpu_str != "None":
+                if instance_type_str.startswith(f"{oci_conf.VM_PREFIX}.A"):
+                    # For ARM cpu, 1*ocpu = 1*vcpu
+                    ocpu_count = round(float(vcpu_str))
+                else:
+                    # For Intel / AMD cpu, 1*ocpu = 2*vcpu
+                    ocpu_count = round(float(vcpu_str) / 2)
+            ocpu_count = 1 if (ocpu_count > 0 and
+                               ocpu_count < 1) else ocpu_count
+
+            machine_shape_config = None
+            if ocpu_count > 0:
+                mem = node_config["MemoryInGbs"]
+                if mem is not None and mem != "None":
+                    machine_shape_config = (oci_adaptor.get_oci().core.models.
+                                            LaunchInstanceShapeConfigDetails(
+                                                ocpus=ocpu_count,
+                                                memory_in_gbs=mem))
+                else:
+                    machine_shape_config = (oci_adaptor.get_oci().core.models.
+                                            LaunchInstanceShapeConfigDetails(
+                                                ocpus=ocpu_count))
+
+            preempitible_config = (oci_adaptor.get_oci(
+            ).core.models.PreemptibleInstanceConfigDetails(
+                preemption_action=oci_adaptor.get_oci().core.models.
+                TerminatePreemptionAction(type="TERMINATE",
+                                          preserve_boot_volume=False))
+                                   if node_config["Preemptible"] else None)
+
+            logger.debug(f"Shape: {instance_type_str}, ocpu: {ocpu_count}")
+            logger.debug(f"Shape config is {machine_shape_config}")
+            logger.debug(f"Spot config is {preempitible_config}")
+
+            vm_tags = {
+                **tags,
+                TAG_RAY_CLUSTER_NAME: self.cluster_name,
+                "sky_spot_flag": str(node_config["Preemptible"]).lower(),
+            }
+            batch_id = datetime.now().strftime("%Y%m%d%H%M%S")
+            node_type = tags[TAG_RAY_NODE_KIND]
+
+            oci_query_helper.subscribe_image(
+                compartment_id=compartment,
+                listing_id=node_config["AppCatalogListingId"],
+                resource_version=node_config["ResourceVersion"],
+                region=self.region,
+            )
+
+            start_time1 = round(time.time() * 1000)
+            for seq in range(1, count + 1):
+                launch_instance_response = oci_adaptor.get_core_client(
+                    self.region, oci_conf.get_profile()
+                ).launch_instance(launch_instance_details=oci_adaptor.get_oci(
+                ).core.models.LaunchInstanceDetails(
+                    availability_domain=node_config["AvailabilityDomain"],
+                    compartment_id=compartment,
+                    shape=instance_type_str,
+                    display_name=
+                    f"{self.cluster_name}_{node_type}_{batch_id}_{seq}",
+                    freeform_tags=vm_tags,
+                    image_id=node_config["ImageId"],
+                    metadata={
+                        "ssh_authorized_keys": node_config["AuthorizedKey"]
+                    },
+                    subnet_id=vcn,
+                    shape_config=machine_shape_config,
+                    preemptible_instance_config=preempitible_config,
+                ))
+
+                new_inst = launch_instance_response.data
+                starting_insts.append({
+                    "inst_id": new_inst.id,
+                    "ad": new_inst.availability_domain,
+                    "compartment": new_inst.compartment_id,
+                    "lifecycle_state": new_inst.lifecycle_state,
+                    "oci_tags": new_inst.freeform_tags,
+                })
+            # end for loop
+
+            launch_new_time = round(time.time() * 1000) - start_time1
+            logger.debug("Time elapsed(Launch): {0} milli-seconds.".format(
+                launch_new_time))
+        # end if count > 0:...
+
+        for ninst in starting_insts:
+            # Waiting for the instance to be RUNNING state
+            get_instance_response = oci_adaptor.get_core_client(
+                self.region, oci_conf.get_profile()).get_instance(
+                    instance_id=ninst["inst_id"])
+            oci_adaptor.get_oci().wait_until(
+                oci_adaptor.get_core_client(self.region,
+                                            oci_conf.get_profile()),
+                get_instance_response,
+                "lifecycle_state",
+                "RUNNING",
+            )
+            ninst["lifecycle_state"] = "RUNNING"
+            self.cached_nodes[ninst["inst_id"]] = self.get_inst_obj(ninst)
+
+        total_time = round(time.time() * 1000) - start_time
+        logger.debug(
+            "Total time elapsed: {0} milli-seconds.".format(total_time))
+
+    def get_inst_obj(self, inst_info):
+        list_vnic_attachments_response = oci_adaptor.get_core_client(
+            self.region, oci_conf.get_profile()).list_vnic_attachments(
+                availability_domain=inst_info["ad"],
+                compartment_id=inst_info["compartment"],
+                instance_id=inst_info["inst_id"],
+            )
+
+        vnic = list_vnic_attachments_response.data[0]
+        get_vnic_response = (oci_adaptor.get_net_client(
+            self.region,
+            oci_conf.get_profile()).get_vnic(vnic_id=vnic.vnic_id).data)
+
+        internal_ip = get_vnic_response.private_ip
+        external_ip = get_vnic_response.public_ip
+        if external_ip is None:
+            external_ip = internal_ip
+
+        return {
+            "id": inst_info["inst_id"],
+            "external_ip": external_ip,
+            "internal_ip": internal_ip,
+            "tags": inst_info["oci_tags"],
+            "status": inst_info["lifecycle_state"],
+        }
+
+    @synchronized
+    @utils.debug_enabled(logger=logger)
+    def set_node_tags(self, node_id, tags):
+        existing_tags = self._get_cached_node(node_id)["tags"]
+        combined_tags = dict(existing_tags, **tags)
+
+        self.cached_nodes[node_id]["tags"] = combined_tags
+        retry_count = 0
+        while retry_count < oci_conf.MAX_RETRY_COUNT:
+            try:
+                oci_adaptor.get_core_client(
+                    self.region, oci_conf.get_profile()).update_instance(
+                        instance_id=node_id,
+                        update_instance_details=oci_adaptor.get_oci().core.
+                        models.UpdateInstanceDetails(
+                            freeform_tags=combined_tags),
+                    )
+                logger.info(f"Tags are well set for node {node_id}")
+                break
+            except Exception as e:
+                retry_count = retry_count + 1
+                wait_seconds = oci_conf.RETRY_INTERVAL_BASE_SECONDS * retry_count
+                logger.warn(
+                    f"Not ready yet, wait {wait_seconds} seconds & retry!")
+                logger.warn(f"Exception message is {str(e)}")
+                time.sleep(wait_seconds)
+
+    @synchronized
+    def terminate_node(self, node_id):
+        """Terminates the specified node."""
+        logger.info(f"terminate_node {node_id}...")
+        node = self._get_cached_node(node_id)
+        if node is None:
+            logger.info(f"The node is not existed: {node_id}..")
+            return  # Node not exists yet.
+
+        logger.debug(f"sky_spot_flag: {node['tags']['sky_spot_flag']}")
+        preemptibleFlag = (True if node and
+                           (str(node["tags"]["sky_spot_flag"]) == "true") else
+                           False)
+
+        if self.cache_stopped_nodes and not preemptibleFlag:
+            logger.info(f"Stopping instance {node_id}"
+                        "(to fully terminate instead, "
+                        "set `cache_stopped_nodes: False` "
+                        "under `provider` in the cluster configuration)")
+            instance_action_response = oci_adaptor.get_core_client(
+                self.region,
+                oci_conf.get_profile()).instance_action(instance_id=node_id,
+                                                        action="STOP")
+            logger.info(
+                f"Stopped the instance {instance_action_response.data.id}")
+            if node_id in self.cached_nodes:
+                self.cached_nodes[node_id]["status"] = "STOPPED"
+            state_word = "Stopped"
+        else:
+            terminate_instance_response = oci_adaptor.get_core_client(
+                self.region, oci_conf.get_profile()).terminate_instance(node_id)
+            logger.debug(terminate_instance_response.data)
+            if node_id in self.cached_nodes:
+                del self.cached_nodes[node_id]
+            state_word = "Terminated"
+
+        logger.info(
+            f"{state_word} {node_id} w/ sky_spot_flag: {preemptibleFlag}.")
+
+    def _get_node(self, node_id):
+        self._get_filtered_nodes({},
+                                 force=True)  # All except for those terminated.
+        return self.cached_nodes.get(node_id, None)
+
+    def _get_cached_node(self, node_id):
+        if node_id in self.cached_nodes:
+            return self.cached_nodes[node_id]
+        return self._get_node(node_id=node_id)
+
+    def stopped_nodes(self, tag_filters):
+        """Return a list of stopped nodes filtered by the specified tags dict."""
+        nodes = self._get_filtered_nodes(tag_filters=tag_filters, force=True)
+        return [
+            v for _, v in nodes.items()
+            if v["status"] in ("STOPPED", "STOPPING")
+        ]
+
+    def running_nodes(self, tag_filters):
+        """Return a list of running node ids filtered by the specified tags dict."""
+        nodes = self._get_filtered_nodes(tag_filters=tag_filters)
+        return [k for k, v in nodes.items() if v["status"] == "RUNNING"]
diff --git a/sky/skylet/providers/oci/query_helper.py b/sky/skylet/providers/oci/query_helper.py
new file mode 100644
index 000000000..0baa8e065
--- /dev/null
+++ b/sky/skylet/providers/oci/query_helper.py
@@ -0,0 +1,365 @@
+"""
+Helper class for some OCI operations methods which needs to be shared/called 
+by multiple places. 
+
+History:
+ - Hysun He (hysun.he@oracle.com) @ Apr, 2023: Initial implementation
+ 
+"""
+
+import logging
+import traceback
+import time
+from datetime import datetime
+import pandas as pd
+import re
+from typing import Optional
+from sky.skylet.providers.oci.config import oci_conf
+from sky.skylet.providers.oci import utils
+from sky.adaptors import oci as oci_adaptor
+
+logger = logging.getLogger(__name__)
+
+
+class oci_query_helper:
+
+    # Call Cloud API to try getting the satisfied nodes.
+    @classmethod
+    @utils.debug_enabled(logger=logger)
+    def query_instances_by_tags(cls, tag_filters, region):
+
+        where_clause_tags = ""
+        for tag_key in tag_filters:
+            if where_clause_tags != "":
+                where_clause_tags += " && "
+
+            tag_value = tag_filters[tag_key]
+            where_clause_tags += (f"(freeformTags.key = '{tag_key}'"
+                                  f" && freeformTags.value = '{tag_value}')")
+
+        qv_str = (f"query instance resources where {where_clause_tags}"
+                  f" && (lifecycleState != 'TERMINATED'"
+                  f" && lifecycleState != 'TERMINATING')")
+
+        qv = oci_adaptor.get_oci(
+        ).resource_search.models.StructuredSearchDetails(
+            query=qv_str,
+            type="Structured",
+            matching_context_type=oci_adaptor.get_oci().resource_search.models.
+            SearchDetails.MATCHING_CONTEXT_TYPE_NONE,
+        )
+
+        list_instances_response = oci_adaptor.get_search_client(
+            region, oci_conf.get_profile()).search_resources(qv)
+        result_set = list_instances_response.data.items
+
+        return result_set
+
+    @classmethod
+    def terminate_instances_by_tags(cls, tag_filters, region) -> int:
+        logger.info(f"Terminate instance by tags: {tag_filters}")
+        insts = cls.query_instances_by_tags(tag_filters, region)
+        fail_count = 0
+        for inst in insts:
+            inst_id = inst.identifier
+            logger.debug(f"Got instance(to be terminated): {inst_id}")
+
+            try:
+                oci_adaptor.get_core_client(
+                    region, oci_conf.get_profile()).terminate_instance(inst_id)
+            except Exception as e:
+                fail_count += 1
+                logger.error(f"Terminate instance failed: {str(e)}\n: {inst}")
+                traceback.print_exc()
+
+        if fail_count == 0:
+            logger.info(f"Instance teardown result: OK")
+        else:
+            logger.warn(f"Instance teardown result: {fail_count} failed!")
+
+        return fail_count
+
+    @classmethod
+    @utils.debug_enabled(logger=logger)
+    def subscribe_image(cls, compartment_id, listing_id, resource_version,
+                        region):
+        if (pd.isna(listing_id) or listing_id.strip() == "None" or
+                listing_id.strip() == "nan"):
+            return
+
+        core_client = oci_adaptor.get_core_client(region,
+                                                  oci_conf.get_profile())
+        try:
+            agreements_response = core_client.get_app_catalog_listing_agreements(
+                listing_id=listing_id, resource_version=resource_version)
+            agreements = agreements_response.data
+
+            core_client.create_app_catalog_subscription(
+                create_app_catalog_subscription_details=oci_adaptor.get_oci(
+                ).core.models.CreateAppCatalogSubscriptionDetails(
+                    compartment_id=compartment_id,
+                    listing_id=listing_id,
+                    listing_resource_version=agreements.
+                    listing_resource_version,
+                    oracle_terms_of_use_link=agreements.
+                    oracle_terms_of_use_link,
+                    time_retrieved=datetime.strptime(
+                        re.sub(
+                            "\d{3}\+\d{2}\:\d{2}",
+                            "Z",
+                            str(agreements.time_retrieved),
+                            0,
+                        ),
+                        "%Y-%m-%d %H:%M:%S.%fZ",
+                    ),
+                    signature=agreements.signature,
+                    eula_link=agreements.eula_link,
+                ))
+        except Exception as e:
+            logger.critical(
+                f"subscribe_image: {listing_id} - {resource_version} ... [Failed]"
+                f"Error message: {str(e)}")
+            raise RuntimeError("ERR: Image subscription error!")
+
+    @classmethod
+    @utils.debug_enabled(logger=logger)
+    def find_compartment(cls, region) -> str:
+        """ If compartment is not configured, we use root compartment """
+        # Try to use the configured one first
+        skypilot_compartment = oci_conf.get_compartment(region)
+        if skypilot_compartment is not None:
+            return skypilot_compartment
+
+        # If not specified, we try to find the one skypilot-compartment
+        root = oci_adaptor.get_oci_config(region)['tenancy']
+        list_compartments_response = oci_adaptor.get_identity_client(
+            region,
+            oci_conf.get_profile()).list_compartments(compartment_id=root,
+                                                      name=oci_conf.COMPARTMENT,
+                                                      lifecycle_state='ACTIVE',
+                                                      limit=1)
+        compartments = list_compartments_response.data
+        if len(compartments) > 0:
+            skypilot_compartment = compartments[0].id
+            return skypilot_compartment
+
+        # Finally, we use root compartment none matches above
+        skypilot_compartment = root
+        return skypilot_compartment
+
+    @classmethod
+    @utils.debug_enabled(logger=logger)
+    def find_create_vcn_subnet(cls, region) -> Optional[str]:
+        """ If sub is not configured, we find/create VCN skypilot_vcn """
+        subnet = oci_conf.get_vcn_subnet(region)
+        if subnet is not None:
+            # User explicitly specified the subnet in sky config.
+            return subnet
+
+        # Try to reuse the skypilot_vcn.
+        net_client = oci_adaptor.get_net_client(region, oci_conf.get_profile())
+        skypilot_compartment = cls.find_compartment(region)
+        list_vcns_response = net_client.list_vcns(
+            compartment_id=skypilot_compartment,
+            display_name=oci_conf.VCN_NAME,
+            lifecycle_state="AVAILABLE")
+        vcns = list_vcns_response.data
+        if len(vcns) > 0:
+            # Found the VCN.
+            skypilot_vcn = vcns[0].id
+            list_subnets_response = net_client.list_subnets(
+                compartment_id=skypilot_compartment,
+                limit=1,
+                vcn_id=skypilot_vcn,
+                display_name=oci_conf.VCN_SUBNET_NAME,
+                lifecycle_state="AVAILABLE")
+            logger.debug(f'Got VCN subnet \n{list_subnets_response.data}')
+            if len(list_subnets_response.data) < 1:
+                logger.error(f'No subnet {oci_conf.VCN_SUBNET_NAME} '
+                             f'found in the VCN {oci_conf.VCN_NAME}')
+                raise RuntimeError(f'VcnSubnetNotFound Error: No subnet '
+                                   f'{oci_conf.VCN_SUBNET_NAME} found in '
+                                   f'the VCN {oci_conf.VCN_NAME}')
+            subnet = list_subnets_response.data[0].id
+            return subnet
+        else:
+            # Create the skypilot_vcn and related resources
+            return cls.create_vcn_subnet(net_client, skypilot_compartment)
+
+    @classmethod
+    @utils.debug_enabled(logger=logger)
+    def create_vcn_subnet(cls, net_client,
+                          skypilot_compartment) -> Optional[str]:
+        try:
+            create_vcn_response = net_client.create_vcn(
+                create_vcn_details=oci_adaptor.get_oci().core.models.
+                CreateVcnDetails(compartment_id=skypilot_compartment,
+                                 cidr_blocks=[oci_conf.VCN_CIDR],
+                                 display_name=oci_conf.VCN_NAME,
+                                 is_ipv6_enabled=False,
+                                 dns_label=oci_conf.VCN_DNS_LABEL))
+            vcn_data = create_vcn_response.data
+            logger.debug(f'Created VCN \n{vcn_data}')
+            skypilot_vcn = vcn_data.id
+            route_table = vcn_data.default_route_table_id
+            security_list = vcn_data.default_security_list_id
+            dhcp_options_id = vcn_data.default_dhcp_options_id
+
+            # Create internet gateway for internet access
+            create_ig_response = net_client.create_internet_gateway(
+                create_internet_gateway_details=oci_adaptor.get_oci(
+                ).core.models.CreateInternetGatewayDetails(
+                    compartment_id=skypilot_compartment,
+                    is_enabled=True,
+                    vcn_id=skypilot_vcn,
+                    display_name=oci_conf.VCN_INTERNET_GATEWAY_NAME))
+            logger.debug(
+                f'Created internet gateway \n{create_ig_response.data}')
+            ig = create_ig_response.data.id
+
+            # Create a public subnet.
+            create_subnet_response = net_client.create_subnet(
+                create_subnet_details=oci_adaptor.get_oci().core.models.
+                CreateSubnetDetails(cidr_block=oci_conf.VCN_SUBNET_CIDR,
+                                    compartment_id=skypilot_compartment,
+                                    vcn_id=skypilot_vcn,
+                                    dhcp_options_id=dhcp_options_id,
+                                    display_name=oci_conf.VCN_SUBNET_NAME,
+                                    prohibit_internet_ingress=False,
+                                    prohibit_public_ip_on_vnic=False,
+                                    route_table_id=route_table,
+                                    security_list_ids=[security_list]))
+            logger.debug(f'Created subnet \n{create_subnet_response.data}')
+            subnet = create_subnet_response.data.id
+
+            list_services_response = net_client.list_services(limit=100)
+            services = [
+                s for s in list_services_response.data
+                if str(s.cidr_block).startswith('all-') and str(s.cidr_block).
+                endswith('-services-in-oracle-services-network')
+            ]
+            if len(services) > 0:
+                # Create service gateway for regional services.
+                create_sg_response = net_client.create_service_gateway(
+                    create_service_gateway_details=oci_adaptor.get_oci(
+                    ).core.models.CreateServiceGatewayDetails(
+                        compartment_id=skypilot_compartment,
+                        services=[
+                            oci_adaptor.get_oci().core.models.
+                            ServiceIdRequestDetails(service_id=services[0].id)
+                        ],
+                        vcn_id=skypilot_vcn))
+                logger.debug(f'Service Gateway: \n{create_sg_response.data}')
+                sg = create_sg_response.data.id
+
+            # Update security list: Allow all traffic in the same subnet
+            update_security_list_response = net_client.update_security_list(
+                security_list_id=security_list,
+                update_security_list_details=oci_adaptor.get_oci(
+                ).core.models.UpdateSecurityListDetails(ingress_security_rules=[
+                    oci_adaptor.get_oci().core.models.IngressSecurityRule(
+                        protocol="6",
+                        source=oci_conf.VCN_CIDR_INTERNET,
+                        is_stateless=False,
+                        source_type="CIDR_BLOCK",
+                        tcp_options=oci_adaptor.get_oci().core.models.
+                        TcpOptions(destination_port_range=oci_adaptor.get_oci().
+                                   core.models.PortRange(max=22, min=22),
+                                   source_port_range=oci_adaptor.get_oci(
+                                   ).core.models.PortRange(max=65535, min=1)),
+                        description="Allow SSH port."),
+                    oci_adaptor.get_oci().core.models.IngressSecurityRule(
+                        protocol="all",
+                        source=oci_conf.VCN_SUBNET_CIDR,
+                        is_stateless=False,
+                        source_type="CIDR_BLOCK",
+                        description="Allow all traffic from/to same subnet."),
+                    oci_adaptor.get_oci().core.models.IngressSecurityRule(
+                        protocol="1",
+                        source=oci_conf.VCN_CIDR_INTERNET,
+                        is_stateless=False,
+                        source_type="CIDR_BLOCK",
+                        icmp_options=oci_adaptor.get_oci(
+                        ).core.models.IcmpOptions(type=3, code=4),
+                        description="ICMP traffic."),
+                    oci_adaptor.get_oci().core.models.IngressSecurityRule(
+                        protocol="1",
+                        source=oci_conf.VCN_CIDR,
+                        is_stateless=False,
+                        source_type="CIDR_BLOCK",
+                        icmp_options=oci_adaptor.get_oci(
+                        ).core.models.IcmpOptions(type=3),
+                        description="ICMP traffic (VCN)."),
+                ]))
+            logger.debug(
+                f'Updated security_list: \n{update_security_list_response.data}'
+            )
+
+            # Update route table: bind to the internet gateway
+            update_route_table_response = net_client.update_route_table(
+                rt_id=route_table,
+                update_route_table_details=oci_adaptor.get_oci(
+                ).core.models.UpdateRouteTableDetails(route_rules=[
+                    oci_adaptor.get_oci().core.models.RouteRule(
+                        network_entity_id=create_ig_response.data.id,
+                        destination='0.0.0.0/0',
+                        destination_type='CIDR_BLOCK',
+                        description='Route table for SkyPilot VCN',
+                        route_type='STATIC')
+                ]))
+            logger.debug(f'Route table: \n{update_route_table_response.data}')
+
+        except oci_adaptor.service_exception() as e:
+            logger.error(f'Create VCN Error: Create new VCN '
+                         f'{oci_conf.VCN_NAME} failed: {str(e)}')
+            # In case of partial success while creating vcn
+            cls.delete_vcn(net_client, skypilot_vcn, subnet, ig, sg)
+            subnet = None
+
+        return subnet
+
+    @classmethod
+    @utils.debug_enabled(logger=logger)
+    def delete_vcn(cls, net_client, skypilot_vcn, skypilot_subnet,
+                   internet_gateway, service_gateway):
+        if skypilot_vcn is None:
+            return  # Nothing to delete
+        try:
+            if internet_gateway is not None:
+                # Delete internet gateway
+                delete_ig_response = net_client.delete_internet_gateway(
+                    ig_id=internet_gateway)
+                logger.debug(f'Deleted internet gateway {internet_gateway}'
+                             f'-{delete_ig_response.data}')
+            if service_gateway is not None:
+                # Delete service gateway
+                delete_sg_response = net_client.delete_service_gateway(
+                    service_gateway_id=service_gateway)
+                logger.debug(f'Deleted service gateway {service_gateway}'
+                             f'-{delete_sg_response.data}')
+            if skypilot_subnet is not None:
+                # Delete subnet
+                delete_subnet_response = net_client.delete_subnet(
+                    subnet_id=skypilot_subnet)
+                logger.debug(f'Deleted subnet {skypilot_subnet}'
+                             f'-{delete_subnet_response.data}')
+            # Delete vcn
+            retry_count = 0
+            while retry_count < oci_conf.MAX_RETRY_COUNT:
+                try:
+                    delete_vcn_response = net_client.delete_vcn(
+                        vcn_id=skypilot_vcn)
+                    logger.debug(
+                        f'Deleted vcn {skypilot_vcn}-{delete_vcn_response.data}'
+                    )
+                    break
+                except oci_adaptor.service_exception() as e:
+                    logger.info(f'Waiting del SG/IG/Subnet finish: {str(e)}')
+                    retry_count = retry_count + 1
+                    if retry_count == oci_conf.MAX_RETRY_COUNT:
+                        raise e
+                    else:
+                        time.sleep(oci_conf.RETRY_INTERVAL_BASE_SECONDS)
+
+        except oci_adaptor.service_exception() as e:
+            logger.error(f'Delete VCN {oci_conf.VCN_NAME} Error: {str(e)}')
diff --git a/sky/skylet/providers/oci/utils.py b/sky/skylet/providers/oci/utils.py
new file mode 100644
index 000000000..6c783a630
--- /dev/null
+++ b/sky/skylet/providers/oci/utils.py
@@ -0,0 +1,21 @@
+from logging import Logger
+from datetime import datetime
+import functools
+
+
+def debug_enabled(logger: Logger):
+
+    def decorate(f):
+
+        @functools.wraps(f)
+        def wrapper(*args, **kwargs):
+            dt_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+            logger.debug(f"{dt_str} Enter {f}, {args}, {kwargs}")
+            try:
+                return f(*args, **kwargs)
+            finally:
+                logger.debug(f"{dt_str} Exit {f}")
+
+        return wrapper
+
+    return decorate
diff --git a/sky/templates/oci-ray.yml.j2 b/sky/templates/oci-ray.yml.j2
new file mode 100644
index 000000000..7dba01c16
--- /dev/null
+++ b/sky/templates/oci-ray.yml.j2
@@ -0,0 +1,134 @@
+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.oci.OCINodeProvider
+  region: {{region}}
+  cache_stopped_nodes: True
+
+auth:
+  ssh_user: ubuntu
+  ssh_private_key: {{ssh_private_key}}
+
+available_node_types:
+  ray_head_default:
+    resources: {}
+    node_config:
+      InstanceType: {{instance_type}}
+      VCPUs: {{cpus}}
+      MemoryInGbs: {{memory}}
+      AvailabilityDomain: {{zone}}
+      ImageId: {{image}}
+      AppCatalogListingId: {{app_catalog_listing_id}}
+      ResourceVersion: {{resource_version}}
+      Preemptible: {{use_spot}}
+      AuthorizedKey: {{ssh_public_key}}
+{% if num_nodes > 1 %}
+  ray_worker_default:
+    min_workers: {{num_nodes - 1}}
+    max_workers: {{num_nodes - 1}}
+    resources: {}
+    node_config:
+      InstanceType: {{instance_type}}
+      VCPUs: {{cpus}}
+      MemoryInGbs: {{memory}}
+      AvailabilityDomain: {{zone}}
+      ImageId: {{image}}
+      AppCatalogListingId: {{app_catalog_listing_id}}
+      ResourceVersion: {{resource_version}}
+      Preemptible: {{use_spot}}
+      AuthorizedKey: {{ssh_public_key}}
+{%- 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:
+  # Disable `unattended-upgrades` to prevent apt-get from hanging. It should be called at the beginning before the process started to avoid being blocked. (This is a temporary fix.)
+  # Create ~/.ssh/config file in case the file does not exist in the image.
+  # Line 'rm ..': there is another installation of pip.
+  # 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;
+    ([ `sudo lshw -class display | grep "NVIDIA Corporation" | wc -l` -gt 0 ]) && (sudo which nvidia-smi > /dev/null || ( sudo apt-get install nvidia-driver-530-open -y && sudo apt-get install nvidia-driver-525-server -y ) || true);
+    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 || (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 install oci;
+    (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 && touch ~/.sudo_as_admin_successful;
+    (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)" && 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');
+    sudo iptables -I INPUT -i ens3 -m state --state ESTABLISHED,RELATED,NEW -j ACCEPT;
+
+# 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): 2
+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_DEDUP_LOGS=0 ray start --disable-usage-stats --head --port={{ray_port}} --dashboard-port={{ray_dashboard_port}} --object-manager-port=8076 --autoscaling-config=~/ray_bootstrap_config.yaml {{"--resources='%s'" % custom_resources if custom_resources}} --temp-dir {{ray_temp_dir}} || exit 1;
+    which prlimit && for id in $(pgrep -f raylet/raylet); do sudo prlimit --nofile=1048576:1048576 --pid=$id || true; done;
+    {{dump_port_command}};
+
+{%- if num_nodes > 1 %}
+worker_start_ray_commands:
+  - ray stop; RAY_SCHEDULER_EVENTS=0 RAY_DEDUP_LOGS=0 ray start --disable-usage-stats --address=$RAY_HEAD_IP:{{ray_port}} --object-manager-port=8076 {{"--resources='%s'" % custom_resources if custom_resources}} --temp-dir {{ray_temp_dir}} || 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
diff --git a/sky/templates/spot-controller.yaml.j2 b/sky/templates/spot-controller.yaml.j2
index fb49f2ace..a507cfc59 100644
--- a/sky/templates/spot-controller.yaml.j2
+++ b/sky/templates/spot-controller.yaml.j2
@@ -29,6 +29,9 @@ setup: |
   #  pip3 install azure-cli==2.31.0 azure-core
   {{google_sdk_installation_commands}}
 
+  echo Check and install OCI SDK
+  pip list | grep oci  > /dev/null 2>&1 || pip install oci 
+
   # Internal: disable logging for manually logging into the spot controller for debugging.
   {% if is_dev %}
   echo 'export SKYPILOT_DEV=1' >> ~/.bashrc


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 3351f669d..3422b8180 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -19,7 +19,7 @@ from typing import List
 #
 # To only run tests for managed spot (without generic tests), use --managed-spot.
 all_clouds_in_smoke_tests = [
-    'aws', 'gcp', 'azure', 'lambda', 'cloudflare', 'ibm', 'scp'
+    'aws', 'gcp', 'azure', 'lambda', 'cloudflare', 'ibm', 'scp', 'oci'
 ]
 default_clouds_to_run = ['gcp', 'azure']
 
@@ -33,7 +33,8 @@ cloud_to_pytest_keyword = {
     'lambda': 'lambda_cloud',
     'cloudflare': 'cloudflare',
     'ibm': 'ibm',
-    'scp': 'scp'
+    'scp': 'scp',
+    'oci': 'oci',
 }
 
 
@@ -183,6 +184,7 @@ def enable_all_clouds(monkeypatch):
         prefix='tmp_backup_config_default', delete=False)
     monkeypatch.setattr('sky.clouds.gcp.GCP_CONFIG_SKY_BACKUP_PATH',
                         config_file_backup.name)
+    monkeypatch.setenv('OCI_CONFIG', config_file_backup.name)
 
 
 @pytest.fixture
diff --git a/tests/test_optimizer_dryruns.py b/tests/test_optimizer_dryruns.py
index 2d288b2c9..fde4125b7 100644
--- a/tests/test_optimizer_dryruns.py
+++ b/tests/test_optimizer_dryruns.py
@@ -61,6 +61,7 @@ def _make_resources(
         prefix='tmp_backup_config_default', delete=False)
     monkeypatch.setattr('sky.clouds.gcp.GCP_CONFIG_SKY_BACKUP_PATH',
                         config_file_backup.name)
+    monkeypatch.setenv('OCI_CONFIG', config_file_backup.name)
 
     # Should create Resources here, since it uses the enabled clouds.
     return sky.Resources(*resources_args, **resources_kwargs)
@@ -437,7 +438,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, GCP and IBM' in str(e.value)
+    assert 'only supported for AWS/GCP/IBM/OCI' in str(e.value)
 
 
 def test_valid_image(monkeypatch):
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index 5fc94303b..1d42b0658 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -815,6 +815,7 @@ def test_scp_logs():
 @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
 @pytest.mark.no_scp  # SCP does not have K80 gpus. Run test_scp_job_queue instead
+@pytest.mark.no_oci  # OCI does not have K80 gpus
 def test_job_queue(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -915,6 +916,7 @@ def test_scp_job_queue():
 @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
 @pytest.mark.no_scp  # SCP does not support num_nodes > 1 yet
+@pytest.mark.no_oci  # OCI Cloud does not have T4 gpus.
 def test_job_queue_multinode(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -1013,6 +1015,7 @@ def test_ibm_job_queue_multinode():
 @pytest.mark.no_lambda_cloud  # Lambda Cloud does not have K80 gpus
 @pytest.mark.no_ibm  # IBM Cloud does not have K80 gpus
 @pytest.mark.no_scp  # SCP does not support num_nodes > 1 yet
+@pytest.mark.no_oci  # OCI Cloud does not have K80 gpus
 def test_multi_echo(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(