Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -190,8 +190,7 @@ oci = [
"dstack[server]",
]
nebius = [
# 0.2.73 breaks sdk backward compatibility: https://github.com/dstackai/dstack/issues/3171
"nebius>=0.2.40,<=0.2.72; python_version >= '3.10'",
"nebius>=0.3.4,<0.4; python_version >= '3.10'",
"dstack[server]",
]
all = [
Expand Down
5 changes: 4 additions & 1 deletion src/dstack/_internal/core/backends/nebius/compute.py
Original file line number Diff line number Diff line change
Expand Up @@ -380,7 +380,10 @@ def _wait_for_instance(sdk: SDK, op: SDKOperation[Operation]) -> None:
)
time.sleep(WAIT_FOR_INSTANCE_UPDATE_INTERVAL)
resources.LOOP.await_(
op.update(per_retry_timeout=resources.REQUEST_TIMEOUT, metadata=resources.REQUEST_MD)
op.update(
per_retry_timeout=resources.REQUEST_TIMEOUT,
auth_options=resources.REQUEST_AUTH_OPTIONS,
)
)


Expand Down
47 changes: 27 additions & 20 deletions src/dstack/_internal/core/backends/nebius/resources.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,6 @@
from tempfile import NamedTemporaryFile
from typing import Dict, Optional

from nebius.aio.authorization.options import options_to_metadata
from nebius.aio.operation import Operation as SDKOperation
from nebius.aio.service_error import RequestError, StatusCode
from nebius.aio.token.renewable import OPTION_RENEW_REQUEST_TIMEOUT, OPTION_RENEW_SYNCHRONOUS
Expand Down Expand Up @@ -66,13 +65,11 @@
LOOP = DaemonEventLoop()
# Pass a timeout to all methods to avoid infinite waiting
REQUEST_TIMEOUT = 10
# Pass REQUEST_MD to all methods to avoid infinite retries in case of invalid credentials
REQUEST_MD = options_to_metadata(
{
OPTION_RENEW_SYNCHRONOUS: "true",
OPTION_RENEW_REQUEST_TIMEOUT: "5",
}
)
# Pass REQUEST_AUTH_OPTIONS to all methods to avoid infinite retries in case of invalid credentials
REQUEST_AUTH_OPTIONS = {
OPTION_RENEW_SYNCHRONOUS: "true",
OPTION_RENEW_REQUEST_TIMEOUT: "5",
}

# disables log messages about errors such as invalid creds or expired timeouts
logging.getLogger("nebius").setLevel(logging.CRITICAL)
Expand Down Expand Up @@ -120,7 +117,9 @@ def wait_for_operation(
if time.monotonic() + interval > deadline:
raise TimeoutError(f"Operation {op.id} wait timeout")
time.sleep(interval)
LOOP.await_(op.update(per_retry_timeout=REQUEST_TIMEOUT, metadata=REQUEST_MD))
LOOP.await_(
op.update(per_retry_timeout=REQUEST_TIMEOUT, auth_options=REQUEST_AUTH_OPTIONS)
)


def get_region_to_project_id_map(
Expand Down Expand Up @@ -156,7 +155,9 @@ def validate_regions(configured: set[str], available: set[str]) -> None:
def list_tenant_projects(sdk: SDK) -> Sequence[Container]:
tenants = LOOP.await_(
TenantServiceClient(sdk).list(
ListTenantsRequest(), per_retry_timeout=REQUEST_TIMEOUT, metadata=REQUEST_MD
ListTenantsRequest(),
per_retry_timeout=REQUEST_TIMEOUT,
auth_options=REQUEST_AUTH_OPTIONS,
)
)
if len(tenants.items) != 1:
Expand All @@ -166,7 +167,7 @@ def list_tenant_projects(sdk: SDK) -> Sequence[Container]:
ProjectServiceClient(sdk).list(
ListProjectsRequest(parent_id=tenant_id, page_size=999),
per_retry_timeout=REQUEST_TIMEOUT,
metadata=REQUEST_MD,
auth_options=REQUEST_AUTH_OPTIONS,
)
)
return projects.items
Expand Down Expand Up @@ -240,7 +241,7 @@ def get_default_subnet(sdk: SDK, project_id: str) -> Subnet:
SubnetServiceClient(sdk).list(
ListSubnetsRequest(parent_id=project_id, page_size=999),
per_retry_timeout=REQUEST_TIMEOUT,
metadata=REQUEST_MD,
auth_options=REQUEST_AUTH_OPTIONS,
)
)
for subnet in subnets.items:
Expand All @@ -267,14 +268,18 @@ def create_disk(
)
with wrap_capacity_errors():
return LOOP.await_(
client.create(request, per_retry_timeout=REQUEST_TIMEOUT, metadata=REQUEST_MD)
client.create(
request, per_retry_timeout=REQUEST_TIMEOUT, auth_options=REQUEST_AUTH_OPTIONS
)
)


def delete_disk(sdk: SDK, disk_id: str) -> None:
LOOP.await_(
DiskServiceClient(sdk).delete(
DeleteDiskRequest(id=disk_id), per_retry_timeout=REQUEST_TIMEOUT, metadata=REQUEST_MD
DeleteDiskRequest(id=disk_id),
per_retry_timeout=REQUEST_TIMEOUT,
auth_options=REQUEST_AUTH_OPTIONS,
)
)

Expand Down Expand Up @@ -325,7 +330,9 @@ def create_instance(
)
with wrap_capacity_errors():
return LOOP.await_(
client.create(request, per_retry_timeout=REQUEST_TIMEOUT, metadata=REQUEST_MD)
client.create(
request, per_retry_timeout=REQUEST_TIMEOUT, auth_options=REQUEST_AUTH_OPTIONS
)
)


Expand All @@ -334,7 +341,7 @@ def get_instance(sdk: SDK, instance_id: str) -> Instance:
InstanceServiceClient(sdk).get(
GetInstanceRequest(id=instance_id),
per_retry_timeout=REQUEST_TIMEOUT,
metadata=REQUEST_MD,
auth_options=REQUEST_AUTH_OPTIONS,
)
)

Expand All @@ -344,7 +351,7 @@ def delete_instance(sdk: SDK, instance_id: str) -> SDKOperation[Operation]:
InstanceServiceClient(sdk).delete(
DeleteInstanceRequest(id=instance_id),
per_retry_timeout=REQUEST_TIMEOUT,
metadata=REQUEST_MD,
auth_options=REQUEST_AUTH_OPTIONS,
)
)

Expand All @@ -358,17 +365,17 @@ def create_cluster(sdk: SDK, name: str, project_id: str, fabric: str) -> SDKOper
spec=GpuClusterSpec(infiniband_fabric=fabric),
),
per_retry_timeout=REQUEST_TIMEOUT,
metadata=REQUEST_MD,
auth_options=REQUEST_AUTH_OPTIONS,
)
)


def delete_cluster(sdk: SDK, cluster_id: str) -> None:
return LOOP.await_(
LOOP.await_(
GpuClusterServiceClient(sdk).delete(
DeleteGpuClusterRequest(id=cluster_id),
per_retry_timeout=REQUEST_TIMEOUT,
metadata=REQUEST_MD,
auth_options=REQUEST_AUTH_OPTIONS,
)
)

Expand Down