skypilot-org
diff --git a/‎docs/source/getting-started/installation.rst
+18 b/‎docs/source/getting-started/installation.rst
+18
diff --git a/‎sky/adaptors/nebius.py
+128-6 b/‎sky/adaptors/nebius.py
+128-6
diff --git a/‎sky/cloud_stores.py
+66 b/‎sky/cloud_stores.py
+66
diff --git a/‎sky/clouds/nebius.py
+4-1 b/‎sky/clouds/nebius.py
+4-1
@@ -584,6 +584,24 @@ To use *Service Account* authentication, follow these steps:
 * The `NEBIUS_IAM_TOKEN` file, if present, will take priority for authentication.
 * Service Accounts are restricted to a single region. Ensure you configure the Service Account for the appropriate region during creation.
 
+Nebius offers `Object Storage <https://nebius.com/services/storage>`_, an S3-compatible object storage without any egress charges.
+SkyPilot can download/upload data to Nebius buckets and mount them as local filesystem on clusters launched by SkyPilot. To set up Nebius support, run:
+
+.. code-block:: shell
+
+  # Install boto
+  pip install boto3
+  # Configure your Nebius Object Storage credentials
+  aws configure --profile nebius
+
+In the prompt, enter your Nebius Access Key ID and Secret Access Key (see `instructions to generate Nebius credentials <https://docs.nebius.com/object-storage/quickstart#env-configure>`_). Select :code:`auto` for the default region and :code:`json` for the default output format.
+
+.. code-block:: bash
+
+  aws configure set aws_access_key_id $NB_ACCESS_KEY_AWS_ID --profile nebius
+  aws configure set aws_secret_access_key $NB_SECRET_ACCESS_KEY --profile nebius
+  aws configure set region eu-west1 --profile nebius
+  aws configure set endpoint_url https://storage.eu-west1.nebius.cloud:443  --profile nebius
 
 Request quotas for first time users
 --------------------------------------
 
@@ -1,7 +1,11 @@
 """Nebius cloud adaptor."""
 import os
+import threading
+from typing import Optional
 
 from sky.adaptors import common
+from sky.utils import annotations
+from sky.utils import ux_utils
 
 NEBIUS_TENANT_ID_FILENAME = 'NEBIUS_TENANT_ID.txt'
 NEBIUS_IAM_TOKEN_FILENAME = 'NEBIUS_IAM_TOKEN.txt'
@@ -12,6 +16,10 @@
 NEBIUS_PROJECT_ID_PATH = '~/.nebius/' + NEBIUS_PROJECT_ID_FILENAME
 NEBIUS_CREDENTIALS_PATH = '~/.nebius/' + NEBIUS_CREDENTIALS_FILENAME
 
+DEFAULT_REGION = 'eu-north1'
+
+NEBIUS_PROFILE_NAME = 'nebius'
+
 MAX_RETRIES_TO_DISK_CREATE = 120
 MAX_RETRIES_TO_INSTANCE_STOP = 120
 MAX_RETRIES_TO_INSTANCE_START = 120
@@ -23,15 +31,27 @@
 POLL_INTERVAL = 5
 
 _iam_token = None
+_sdk = None
 _tenant_id = None
 _project_id = None
 
+_IMPORT_ERROR_MESSAGE = ('Failed to import dependencies for Nebius AI Cloud.'
+                         'Try pip install "skypilot[nebius]"')
+
 nebius = common.LazyImport(
     'nebius',
-    import_error_message='Failed to import dependencies for Nebius AI Cloud. '
-    'Try running: pip install "skypilot[nebius]"',
+    import_error_message=_IMPORT_ERROR_MESSAGE,
     # https://github.com/grpc/grpc/issues/37642 to avoid spam in console
     set_loggers=lambda: os.environ.update({'GRPC_VERBOSITY': 'NONE'}))
+boto3 = common.LazyImport('boto3', import_error_message=_IMPORT_ERROR_MESSAGE)
+botocore = common.LazyImport('botocore',
+                             import_error_message=_IMPORT_ERROR_MESSAGE)
+
+_LAZY_MODULES = (boto3, botocore, nebius)
+_session_creation_lock = threading.RLock()
+_INDENT_PREFIX = '    '
+NAME = 'Nebius'
+SKY_CHECK_NAME = 'Nebius (for Nebius Object Storae)'
 
 
 def request_error():
@@ -104,7 +124,109 @@ def get_tenant_id():
 
 
 def sdk():
-    if get_iam_token() is not None:
-        return nebius.sdk.SDK(credentials=get_iam_token())
-    return nebius.sdk.SDK(
-        credentials_file_name=os.path.expanduser(NEBIUS_CREDENTIALS_PATH))
+    global _sdk
+    if _sdk is None:
+        if get_iam_token() is not None:
+            _sdk = nebius.sdk.SDK(credentials=get_iam_token())
+            return _sdk
+        _sdk = nebius.sdk.SDK(
+            credentials_file_name=os.path.expanduser(NEBIUS_CREDENTIALS_PATH))
+    return _sdk
+
+
+def get_nebius_credentials(boto3_session):
+    """Gets the Nebius credentials from the boto3 session object.
+
+    Args:
+        boto3_session: The boto3 session object.
+    Returns:
+        botocore.credentials.ReadOnlyCredentials object with the R2 credentials.
+    """
+    nebius_credentials = boto3_session.get_credentials()
+    if nebius_credentials is None:
+        with ux_utils.print_exception_no_traceback():
+            raise ValueError('Nebius credentials not found. Run '
+                             '`sky check` to verify credentials are '
+                             'correctly set up.')
+    return nebius_credentials.get_frozen_credentials()
+
+
+# lru_cache() is thread-safe and it will return the same session object
+# for different threads.
+# Reference: https://docs.python.org/3/library/functools.html#functools.lru_cache # pylint: disable=line-too-long
+@annotations.lru_cache(scope='global')
+def session():
+    """Create an AWS session."""
+    # Creating the session object is not thread-safe for boto3,
+    # so we add a reentrant lock to synchronize the session creation.
+    # Reference: https://github.com/boto/boto3/issues/1592
+    # However, the session object itself is thread-safe, so we are
+    # able to use lru_cache() to cache the session object.
+    with _session_creation_lock:
+        session_ = boto3.session.Session(profile_name=NEBIUS_PROFILE_NAME)
+    return session_
+
+
+@annotations.lru_cache(scope='global')
+def resource(resource_name: str, region: str = DEFAULT_REGION, **kwargs):
+    """Create a Nebius resource.
+
+    Args:
+        resource_name: Nebius resource name (e.g., 's3').
+        kwargs: Other options.
+    """
+    # Need to use the resource retrieved from the per-thread session
+    # to avoid thread-safety issues (Directly creating the client
+    # with boto3.resource() is not thread-safe).
+    # Reference: https://stackoverflow.com/a/59635814
+
+    session_ = session()
+    nebius_credentials = get_nebius_credentials(session_)
+    endpoint = create_endpoint(region)
+
+    return session_.resource(
+        resource_name,
+        endpoint_url=endpoint,
+        aws_access_key_id=nebius_credentials.access_key,
+        aws_secret_access_key=nebius_credentials.secret_key,
+        region_name=region,
+        **kwargs)
+
+
+@annotations.lru_cache(scope='global')
+def client(service_name: str, region):
+    """Create an Nebius client of a certain service.
+
+    Args:
+        service_name: Nebius service name (e.g., 's3').
+        kwargs: Other options.
+    """
+    # Need to use the client retrieved from the per-thread session
+    # to avoid thread-safety issues (Directly creating the client
+    # with boto3.client() is not thread-safe).
+    # Reference: https://stackoverflow.com/a/59635814
+
+    session_ = session()
+    nebius_credentials = get_nebius_credentials(session_)
+    endpoint = create_endpoint(region)
+
+    return session_.client(service_name,
+                           endpoint_url=endpoint,
+                           aws_access_key_id=nebius_credentials.access_key,
+                           aws_secret_access_key=nebius_credentials.secret_key,
+                           region_name=region)
+
+
+@common.load_lazy_modules(_LAZY_MODULES)
+def botocore_exceptions():
+    """AWS botocore exception."""
+    # pylint: disable=import-outside-toplevel
+    from botocore import exceptions
+    return exceptions
+
+
+def create_endpoint(region: Optional[str] = DEFAULT_REGION) -> str:
+    """Reads accountid necessary to interact with Nebius Object Storage"""
+    if region is None:
+        region = DEFAULT_REGION
+    return f'https://storage.{region}.nebius.cloud:443'
@@ -19,6 +19,7 @@
 from sky.adaptors import azure
 from sky.adaptors import cloudflare
 from sky.adaptors import ibm
+from sky.adaptors import nebius
 from sky.adaptors import oci
 from sky.clouds import gcp
 from sky.data import data_utils
@@ -543,6 +544,70 @@ def make_sync_file_command(self, source: str, destination: str) -> str:
         return download_via_ocicli
 
 
+class NebiusCloudStorage(CloudStorage):
+    """Nebius Cloud Storage."""
+
+    # List of commands to install AWS CLI
+    _GET_AWSCLI = [
+        'aws --version >/dev/null 2>&1 || '
+        f'{constants.SKY_UV_PIP_CMD} install awscli',
+    ]
+
+    def is_directory(self, url: str) -> bool:
+        """Returns whether nebius 'url' is a directory.
+
+        In cloud object stores, a "directory" refers to a regular object whose
+        name is a prefix of other objects.
+        """
+        nebius_s3 = nebius.resource('s3')
+        bucket_name, path = data_utils.split_nebius_path(url)
+        bucket = nebius_s3.Bucket(bucket_name)
+
+        num_objects = 0
+        for obj in bucket.objects.filter(Prefix=path):
+            num_objects += 1
+            if obj.key == path:
+                return False
+            # If there are more than 1 object in filter, then it is a directory
+            if num_objects == 3:
+                return True
+
+        # A directory with few or no items
+        return True
+
+    def make_sync_dir_command(self, source: str, destination: str) -> str:
+        """Downloads using AWS CLI."""
+        # AWS Sync by default uses 10 threads to upload files to the bucket.
+        # To increase parallelism, modify max_concurrent_requests in your
+        # aws config file (Default path: ~/.aws/config).
+        endpoint_url = nebius.create_endpoint()
+        assert 'nebius://' in source, 'nebius:// is not in source'
+        source = source.replace('nebius://', 's3://')
+        download_via_awscli = (f'{constants.SKY_REMOTE_PYTHON_ENV}/bin/aws s3 '
+                               'sync --no-follow-symlinks '
+                               f'{source} {destination} '
+                               f'--endpoint {endpoint_url} '
+                               f'--profile={nebius.NEBIUS_PROFILE_NAME}')
+
+        all_commands = list(self._GET_AWSCLI)
+        all_commands.append(download_via_awscli)
+        return ' && '.join(all_commands)
+
+    def make_sync_file_command(self, source: str, destination: str) -> str:
+        """Downloads a file using AWS CLI."""
+        endpoint_url = nebius.create_endpoint()
+        assert 'nebius://' in source, 'nebius:// is not in source'
+        source = source.replace('nebius://', 's3://')
+        download_via_awscli = (f'{constants.SKY_REMOTE_PYTHON_ENV}/bin/aws s3 '
+                               f'cp {source} {destination} '
+                               f'--endpoint {endpoint_url} '
+                               f'--profile={nebius.NEBIUS_PROFILE_NAME}')
+
+        all_commands = list(self._GET_AWSCLI)
+        all_commands.append(download_via_awscli)
+        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)
@@ -559,6 +624,7 @@ def get_storage_from_path(url: str) -> CloudStorage:
     'r2': R2CloudStorage(),
     'cos': IBMCosCloudStorage(),
     'oci': OciCloudStorage(),
+    'nebius': NebiusCloudStorage(),
     # 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`
 
@@ -279,10 +279,13 @@ def check_credentials(cls) -> Tuple[bool, Optional[str]]:
         return True, None
 
     def get_credential_file_mounts(self) -> Dict[str, str]:
-        return {
+        credential_file_mounts = {
             f'~/.nebius/{filename}': f'~/.nebius/{filename}'
             for filename in _CREDENTIAL_FILES
         }
+        credential_file_mounts['~/.aws/credentials'] = '~/.aws/credentials'
+
+        return credential_file_mounts
 
     @classmethod
     def get_current_user_identity(cls) -> Optional[List[str]]: