[OCI] Support OCI Object Storage

examples/oci/dataset-mount.yaml

@@ -0,0 +1,41 @@
+name: cpu-task1
+
+resources:
+  # Optional; if left out, automatically pick the cheapest cloud.
+  cloud: oci
+  region: us-sanjose-1
+  #zone: AP-SEOUL-1-AD-1
+  #instance_type: VM.Standard.E4.Flex$_2_8
+  cpus: 2
+  #image_id: skypilot:cpu-ubuntu-2004
+  disk_size: 256
+  disk_tier: medium
+  use_spot: False
+
+file_mounts:
+  # Mount an existing oci bucket
+  /datasets-storage:
+    source: oci://skybucket5
+    mode: MOUNT  # Either MOUNT or COPY. Optional.
+
+# 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 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."

examples/oci/dataset-upload-and-mount.yaml

@@ -0,0 +1,43 @@
+name: cpu-task1
+
+resources:
+  # Optional; if left out, automatically pick the cheapest cloud.
+  cloud: oci
+  region: us-sanjose-1
+  #zone: AP-SEOUL-1-AD-1
+  #instance_type: VM.Standard.E4.Flex$_2_8
+  cpus: 2
+  #image_id: skypilot:cpu-ubuntu-2004
+  disk_size: 256
+  disk_tier: medium
+  use_spot: False
+
+file_mounts:
+  /datasets-storage:
+    name: skybucket5  # Name of storage, optional when source is bucket URI
+    source: ['~/Hysun/dataset']  # Source path, can be local or bucket URL. Optional, do not specify to create an empty bucket.
+    store: oci  # E.g 'oci', 's3', 'gcs'...; default: None. Optional.
+    persistent: True  # Defaults to True; can be set to false. Optional.
+    mode: MOUNT  # Either MOUNT or COPY. Optional.
+
+# 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 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."

examples/oci/oci-mounts.yaml

@@ -0,0 +1,26 @@
+resources:
+  cloud: oci
+
+file_mounts:
+  ~/tmpfile: ~/tmpfile
+  ~/a/b/c/tmpfile: ~/tmpfile
+  /tmp/workdir: ~/tmp-workdir
+
+  /mydir:
+    name: skybucket
+    source: ['~/tmp-workdir']
+    store: oci
+    mode: MOUNT
+
+setup: |
+  echo "*** Setup ***"
+
+run: |
+  echo "*** Run ***"
+
+  ls -lthr ~/tmpfile
+  ls -lthr ~/a/b/c
+  echo hi >> /tmp/workdir/new_file
+  ls -lthr /tmp/workdir
+
+  ls -lthr /mydir

sky/adaptors/oci.py

@@ -2,8 +2,10 @@
 
 import logging
 import os
+from typing import List
 
 from sky.adaptors import common
+from sky.clouds.utils import oci_utils
 
 # Suppress OCI circuit breaker logging before lazy import, because
 # oci modules prints additional message during imports, i.e., the
@@ -30,10 +32,16 @@ def get_config_file() -> str:
 
 def get_oci_config(region=None, profile='DEFAULT'):
     conf_file_path = get_config_file()
+    if not profile or profile == 'DEFAULT':
+        config_profile = oci_utils.oci_config.get_profile()
+    else:
+        config_profile = profile
+
     oci_config = oci.config.from_file(file_location=conf_file_path,
-                                      profile_name=profile)
+                                      profile_name=config_profile)
     if region is not None:
         oci_config['region'] = region
+
     return oci_config
 
 
@@ -54,6 +62,28 @@ def get_identity_client(region=None, profile='DEFAULT'):
     return oci.identity.IdentityClient(get_oci_config(region, profile))
 
 
+def get_object_storage_client(region=None, profile='DEFAULT'):
+    return oci.object_storage.ObjectStorageClient(
+        get_oci_config(region, profile))
+
+
+def goto_oci_cli_venv() -> List:
+    # Create a specfic venv for oci-cli due to its dependancy conflict
+    # with runpod (on 'click' version)
+    # pylint: disable=line-too-long
+    cmds = [
+        'conda info --envs | grep "sky-oci-cli-env" || conda create -n sky-oci-cli-env python=3.10 -y',
+        '. $(conda info --base 2> /dev/null)/etc/profile.d/conda.sh > /dev/null 2>&1 || true',
+        'conda activate sky-oci-cli-env', 'pip install oci-cli',
+        'export OCI_CLI_SUPPRESS_FILE_PERMISSIONS_WARNING=True'
+    ]
+    return cmds
+
+
+def leave_oci_cli_venv() -> str:
+    return 'conda deactivate'
+
+
 def service_exception():
     """OCI service exception."""
     return oci.exceptions.ServiceError

sky/cloud_stores.py

@@ -7,6 +7,7 @@
 * Better interface.
 * Better implementation (e.g., fsspec, smart_open, using each cloud's SDK).
 """
+import os
 import shlex
 import subprocess
 import time
@@ -18,6 +19,7 @@
 from sky.adaptors import azure
 from sky.adaptors import cloudflare
 from sky.adaptors import ibm
+from sky.adaptors import oci
 from sky.clouds import gcp
 from sky.data import data_utils
 from sky.data.data_utils import Rclone
@@ -470,6 +472,72 @@ def make_sync_file_command(self, source: str, destination: str) -> str:
         return self.make_sync_dir_command(source, destination)
 
 
+class OciCloudStorage(CloudStorage):
+    """OCI Cloud Storage."""
+
+    def is_directory(self, url: str) -> bool:
+        """Returns whether OCI 'url' is a directory.
+        In cloud object stores, a "directory" refers to a regular object whose
+        name is a prefix of other objects.
+        """
+        bucket_name, path = data_utils.split_oci_path(url)
+
+        client = oci.get_object_storage_client()
+        namespace = client.get_namespace(
+            compartment_id=oci.get_oci_config()['tenancy']).data
+
+        objects = client.list_objects(namespace_name=namespace,
+                                      bucket_name=bucket_name,
+                                      prefix=path).data.objects
+
+        if len(objects) == 0:
+            # A directory with few or no items
+            return True
+
+        if len(objects) > 1:
+            # A directory with more than 1 items
+            return True
+
+        object_name = objects[0].name
+        if path.endswith(object_name):
+            # An object path
+            return False
+
+        # A directory with only 1 item
+        return True
+
+    def make_sync_dir_command(self, source: str, destination: str) -> str:
+        """Downloads using OCI CLI."""
+        bucket_name, path = data_utils.split_oci_path(source)
+
+        download_via_ocicli = (f'oci os object sync --no-follow-symlinks '
+                               f'--bucket-name {bucket_name} '
+                               f'--prefix "{path}" --dest-dir "{destination}"')
+
+        all_commands = oci.goto_oci_cli_venv()
+        all_commands.append(download_via_ocicli)
+        all_commands.append(oci.leave_oci_cli_venv())
+        return ' && '.join(all_commands)
+
+    def make_sync_file_command(self, source: str, destination: str) -> str:
+        """Downloads a file using OCI CLI."""
+        bucket_name, path = data_utils.split_oci_path(source)
+        filename = os.path.basename(path)
+
+        if destination.endswith('/'):
+            destination = f'{destination}{filename}'
+        else:
+            destination = f'{destination}/{filename}'
+
+        download_via_ocicli = (f'oci os object get --bucket-name {bucket_name} '
+                               f'--name "{path}" --file "{destination}"')
+
+        all_commands = oci.goto_oci_cli_venv()
+        all_commands.append(download_via_ocicli)
+        all_commands.append(oci.leave_oci_cli_venv())
+        return ' && '.join(all_commands)
+
+
 def get_storage_from_path(url: str) -> CloudStorage:
     """Returns a CloudStorage by identifying the scheme:// in a URL."""
     result = urllib.parse.urlsplit(url)
@@ -485,6 +553,7 @@ def get_storage_from_path(url: str) -> CloudStorage:
     's3': S3CloudStorage(),
     'r2': R2CloudStorage(),
     'cos': IBMCosCloudStorage(),
+    'oci': OciCloudStorage(),
     # TODO: This is a hack, as Azure URL starts with https://, we should
     # refactor the registry to be able to take regex, so that Azure blob can
     # be identified with `https://(.*?)\.blob\.core\.windows\.net`

sky/data/data_transfer.py

@@ -200,3 +200,77 @@ def _add_bucket_iam_member(bucket_name: str, role: str, member: str) -> None:
     bucket.set_iam_policy(policy)
 
     logger.debug(f'Added {member} with role {role} to {bucket_name}.')
+
+
+def s3_to_oci(s3_bucket_name: str, oci_bucket_name: str) -> None:
+    """Creates a one-time transfer from Amazon S3 to OCI Object Storage.
+    Args:
+      s3_bucket_name: str; Name of the Amazon S3 Bucket
+      oci_bucket_name: str; Name of the OCI Bucket
+    """
+    # TODO(HysunHe): Implement sync with other clouds (s3, gs)
+    raise NotImplementedError('Moving data directly from S3 to OCI bucket '
+                              'is currently not supported. Please specify '
+                              'a local source for the storage object.')
+
+
+def gcs_to_oci(gs_bucket_name: str, oci_bucket_name: str) -> None:
+    """Creates a one-time transfer from Google Cloud Storage to
+    OCI Object Storage.
+    Args:
+      gs_bucket_name: str; Name of the Google Cloud Storage Bucket
+      oci_bucket_name: str; Name of the OCI Bucket
+    """
+    # TODO(HysunHe): Implement sync with other clouds (s3, gs)
+    raise NotImplementedError('Moving data directly from GCS to OCI bucket '
+                              'is currently not supported. Please specify '
+                              'a local source for the storage object.')
+
+
+def r2_to_oci(r2_bucket_name: str, oci_bucket_name: str) -> None:
+    """Creates a one-time transfer from Cloudflare R2 to OCI Bucket.
+    Args:
+      r2_bucket_name: str; Name of the Cloudflare R2 Bucket
+      oci_bucket_name: str; Name of the OCI Bucket
+    """
+    raise NotImplementedError(
+        'Moving data directly from Cloudflare R2 to OCI '
+        'bucket is currently not supported. Please specify '
+        'a local source for the storage object.')
+
+
+def oci_to_gcs(oci_bucket_name: str, gs_bucket_name: str) -> None:
+    """Creates a one-time transfer from OCI Object Storage to
+    Google Cloud Storage.
+    Args:
+      oci_bucket_name: str; Name of the OCI Bucket
+      gs_bucket_name: str; Name of the Google Cloud Storage Bucket
+    """
+    # TODO(HysunHe): Implement sync with other clouds (s3, gs)
+    raise NotImplementedError('Moving data directly from OCI to GCS bucket '
+                              'is currently not supported. Please specify '
+                              'a local source for the storage object.')
+
+
+def oci_to_s3(oci_bucket_name: str, gs_bucket_name: str) -> None:
+    """Creates a one-time transfer from OCI Object Storage to Amazon S3.
+    Args:
+      oci_bucket_name: str; Name of the OCI Bucket
+      s3_bucket_name: str; Name of the Amazon S3 Bucket
+    """
+    # TODO(HysunHe): Implement sync with other clouds (s3, gs)
+    raise NotImplementedError('Moving data directly from OCI to S3 bucket '
+                              'is currently not supported. Please specify '
+                              'a local source for the storage object.')
+
+
+def oci_to_r2(oci_bucket_name: str, r2_bucket_name: str) -> None:
+    """Creates a one-time transfer from OCI Object Storage to
+    Cloudflare R2 Bucket.
+    Args:
+      oci_bucket_name: str; Name of the OCI Bucket
+      r2_bucket_name: str; Name of the Cloudflare R2 Bucket
+    """
+    raise NotImplementedError('Moving data directly from OCI to Cloudflare '
+                              'R2 bucket is currently not supported. Please '
+                              'specify a local source for the storage object.')

sky/data/data_utils.py

@@ -737,3 +737,28 @@ def _remove_bucket_profile_rclone(bucket_name: str,
                 lines_to_keep.append(line)
 
         return lines_to_keep
+
+
+def split_oci_path(oci_path: str) -> Tuple[str, str]:
+    """Splits OCI Path into Bucket name and Relative Path to Bucket
+    Args:
+      oci_path: str; OCI Path, e.g. oci://imagenet/train/
+    """
+    path_parts = oci_path.replace('oci://', '').split('/')
+    bucket = path_parts.pop(0)
+    key = '/'.join(path_parts)
+    return bucket, key
+
+
+def verify_oci_bucket(name: str) -> bool:
+    """Helper method that checks if the OCI bucket exists
+    This method is mainly used by other cloud stores to check the
+    existence of an OCI bucket when it is specified as source. However,
+    We don't verify the existence of OCI bucket because moving data
+    directly between other cloud buckets and OCI buckets is currently
+    not supported.
+    Args:
+      name: str; Name of OCI Bucket (without oci:// prefix)
+    """
+    logger.debug(f'verify_oci_bucket: {name}')
+    return True