Support Nebius tenancies with multiple projects (#2575)

As it turns out, Nebius tenancies can have more
than one project per region, this feature is just
not public yet.

Address such tenancies in `dstack`:
- Allow configuring projects in the backend
  config.
- When projects are not configured and `dstack`
  cannot detect the default project, return a
  backend configuration error, as this most likely
  means the backend needs additional
  configuration.

Author: jvstme

docs/docs/concepts/backends.md

@@ -647,6 +647,25 @@ projects:
 
     </div>
 
+??? info "Projects"
+    If you have multiple projects per region, specify which ones to use, at most one per region.
+
+    <div editor-title="~/.dstack/server/config.yml">
+
+    ```yaml
+    type: nebius
+    projects:
+    - project-e00jt6t095t1ahrg4re30
+    - project-e01iahuh3cklave4ao1nv
+    creds:
+      type: service_account
+      service_account_id: serviceaccount-e00dhnv9ftgb3cqmej
+      public_key_id: publickey-e00ngaex668htswqy4
+      private_key_file: ~/path/to/key.pem
+    ```
+
+    </div>
+
 !!! info "Python version"
     Nebius is only supported if `dstack server` is running on Python 3.10 or higher.
 

src/dstack/_internal/core/backends/nebius/compute.py

@@ -86,7 +86,11 @@ def _sdk(self) -> SDK:
 
     @cached_property
     def _region_to_project_id(self) -> dict[str, str]:
-        return resources.get_region_to_project_id_map(self._sdk)
+        return resources.get_region_to_project_id_map(
+            self._sdk,
+            configured_regions=self.config.regions,
+            configured_project_ids=self.config.projects,
+        )
 
     def _get_subnet_id(self, region: str) -> str:
         if region not in self._subnet_id_cache:
@@ -100,7 +104,7 @@ def get_offers(
     ) -> List[InstanceOfferWithAvailability]:
         offers = get_catalog_offers(
             backend=BackendType.NEBIUS,
-            locations=self.config.regions or list(self._region_to_project_id),
+            locations=list(self._region_to_project_id),
             requirements=requirements,
             extra_filter=_supported_instances,
             configurable_disk_size=CONFIGURABLE_DISK_SIZE,

src/dstack/_internal/core/backends/nebius/configurator.py

@@ -29,21 +29,15 @@ def validate_config(self, config: NebiusBackendConfigWithCreds, default_creds_en
         assert isinstance(config.creds, NebiusServiceAccountCreds)
         try:
             sdk = resources.make_sdk(config.creds)
-            available_regions = set(resources.get_region_to_project_id_map(sdk))
+            # check that it's possible to build the projects map with configured settings
+            resources.get_region_to_project_id_map(
+                sdk, configured_regions=config.regions, configured_project_ids=config.projects
+            )
         except (ValueError, RequestError) as e:
             raise_invalid_credentials_error(
                 fields=[["creds"]],
                 details=str(e),
             )
-        if invalid_regions := set(config.regions or []) - available_regions:
-            raise_invalid_credentials_error(
-                fields=[["regions"]],
-                details=(
-                    f"Configured regions {invalid_regions} do not exist in this Nebius tenancy."
-                    " Omit `regions` to use all regions or select some of the available regions:"
-                    f" {available_regions}"
-                ),
-            )
 
     def create_backend(
         self, project_name: str, config: NebiusBackendConfigWithCreds

src/dstack/_internal/core/backends/nebius/models.py

@@ -5,6 +5,8 @@
 from dstack._internal.core.backends.base.models import fill_data
 from dstack._internal.core.models.common import CoreModel
 
+DEFAULT_PROJECT_NAME_PREFIX = "default-project"
+
 
 class NebiusServiceAccountCreds(CoreModel):
     type: Annotated[Literal["service_account"], Field(description="The type of credentials")] = (
@@ -70,9 +72,20 @@ class NebiusBackendConfig(CoreModel):
         Literal["nebius"],
         Field(description="The type of backend"),
     ] = "nebius"
+    projects: Annotated[
+        Optional[list[str]],
+        Field(
+            description=(
+                "The list of allowed Nebius project IDs."
+                " Omit to use the default project in each region."
+                " The project is considered default if it is the only project in the region"
+                f" or if its name starts with `{DEFAULT_PROJECT_NAME_PREFIX}`"
+            )
+        ),
+    ] = None
     regions: Annotated[
         Optional[list[str]],
-        Field(description="The list of Nebius regions. Omit to use all regions"),
+        Field(description="The list of allowed Nebius regions. Omit to allow all regions"),
     ] = None
 
 

src/dstack/_internal/core/backends/nebius/resources.py

@@ -2,9 +2,10 @@
 import time
 from collections import defaultdict
 from collections.abc import Container as ContainerT
-from collections.abc import Generator
+from collections.abc import Generator, Iterable, Sequence
 from contextlib import contextmanager
 from tempfile import NamedTemporaryFile
+from typing import Optional
 
 from nebius.aio.authorization.options import options_to_metadata
 from nebius.aio.operation import Operation as SDKOperation
@@ -40,7 +41,11 @@
 from nebius.api.nebius.vpc.v1 import ListSubnetsRequest, Subnet, SubnetServiceClient
 from nebius.sdk import SDK
 
-from dstack._internal.core.backends.nebius.models import NebiusServiceAccountCreds
+from dstack._internal.core.backends.base.configurator import raise_invalid_credentials_error
+from dstack._internal.core.backends.nebius.models import (
+    DEFAULT_PROJECT_NAME_PREFIX,
+    NebiusServiceAccountCreds,
+)
 from dstack._internal.core.errors import BackendError, NoCapacityError
 from dstack._internal.utils.event_loop import DaemonEventLoop
 from dstack._internal.utils.logging import get_logger
@@ -110,7 +115,37 @@ def wait_for_operation(
         LOOP.await_(op.update(timeout=REQUEST_TIMEOUT, metadata=REQUEST_MD))
 
 
-def get_region_to_project_id_map(sdk: SDK) -> dict[str, str]:
+def get_region_to_project_id_map(
+    sdk: SDK, configured_regions: Optional[list[str]], configured_project_ids: Optional[list[str]]
+) -> dict[str, str]:
+    """Validate backend settings and build region->project_id map"""
+
+    projects = list_tenant_projects(sdk)
+    if configured_regions:
+        validate_regions(
+            configured=set(configured_regions), available={p.status.region for p in projects}
+        )
+    if configured_project_ids is not None:
+        return _get_region_to_configured_project_id_map(
+            projects, configured_project_ids, configured_regions
+        )
+    else:
+        return _get_region_to_default_project_id_map(projects, configured_regions)
+
+
+def validate_regions(configured: set[str], available: set[str]) -> None:
+    if invalid := set(configured) - available:
+        raise_invalid_credentials_error(
+            fields=[["regions"]],
+            details=(
+                f"Configured regions {invalid} do not exist in this Nebius tenancy."
+                " Omit `regions` to use all regions or select some of the available regions:"
+                f" {available}"
+            ),
+        )
+
+
+def list_tenant_projects(sdk: SDK) -> Sequence[Container]:
     tenants = LOOP.await_(
         TenantServiceClient(sdk).list(
             ListTenantsRequest(), timeout=REQUEST_TIMEOUT, metadata=REQUEST_MD
@@ -126,26 +161,72 @@ def get_region_to_project_id_map(sdk: SDK) -> dict[str, str]:
             metadata=REQUEST_MD,
         )
     )
+    return projects.items
+
+
+def _get_region_to_default_project_id_map(
+    all_tenant_projects: Iterable[Container], configured_regions: Optional[list[str]]
+) -> dict[str, str]:
     region_to_projects: defaultdict[str, list[Container]] = defaultdict(list)
-    for project in projects.items:
+    for project in all_tenant_projects:
         region_to_projects[project.status.region].append(project)
     region_to_project_id = {}
     for region, region_projects in region_to_projects.items():
+        if configured_regions and region not in configured_regions:
+            continue
         if len(region_projects) != 1:
-            # Currently, there can only be one project per region.
-            # This condition is implemented just in case Nebius suddenly allows more projects.
             region_projects = [
-                p for p in region_projects if p.metadata.name.startswith("default-project")
+                p
+                for p in region_projects
+                if p.metadata.name.startswith(DEFAULT_PROJECT_NAME_PREFIX)
             ]
             if len(region_projects) != 1:
-                logger.warning(
-                    "Could not find the default project in region %s, tenant %s", region, tenant_id
+                raise_invalid_credentials_error(
+                    ["regions"],
+                    (
+                        f"Could not find the default project in region {region}."
+                        " Consider setting the `projects` property in backend settings"
+                    ),
                 )
-                continue
         region_to_project_id[region] = region_projects[0].metadata.id
     return region_to_project_id
 
 
+def _get_region_to_configured_project_id_map(
+    all_tenant_projects: Iterable[Container],
+    configured_project_ids: list[str],
+    configured_regions: Optional[list[str]],
+) -> dict[str, str]:
+    project_id_to_project = {p.metadata.id: p for p in all_tenant_projects}
+    region_to_project_id = {}
+    for project_id in configured_project_ids:
+        project = project_id_to_project.get(project_id)
+        if project is None:
+            raise_invalid_credentials_error(
+                ["projects"],
+                f"Configured project ID {project_id!r} not found in this Nebius tenancy",
+            )
+        duplicate_project_id = region_to_project_id.get(project.status.region)
+        if duplicate_project_id:
+            raise_invalid_credentials_error(
+                ["projects"],
+                (
+                    f"Configured projects {project_id} and {duplicate_project_id}"
+                    f" both belong to the same region {project.status.region}."
+                    " Only one project per region is allowed"
+                ),
+            )
+        region_to_project_id[project.status.region] = project_id
+    if configured_regions:
+        # only filter by region after validating all project IDs
+        return {
+            region: project_id
+            for region, project_id in region_to_project_id.items()
+            if region in configured_regions
+        }
+    return region_to_project_id
+
+
 def get_default_subnet(sdk: SDK, project_id: str) -> Subnet:
     subnets = LOOP.await_(
         SubnetServiceClient(sdk).list(