Support reuse-only cloud fleet exports

mkdocs/docs/concepts/exports.md

@@ -17,7 +17,11 @@ Once an export is created, the importer projects can see the exported resources
 for running tasks, dev environments, and services. Imported resources appear with a project prefix
 (e.g., `team-a/my-fleet`) to distinguish them from the project's own resources.
 
-Currently, [SSH fleets](fleets.md#ssh-fleets) and [gateways](gateways.md) can be exported.
+Currently, [fleets](fleets.md) and [gateways](gateways.md) can be exported.
+
+??? info "Limitations"
+    - Importer projects can only use the existing instances of imported [backend fleets](fleets.md#backend-fleets). New instances are not provisioned for importers' runs.
+    - Runs that mount [network volumes](volumes.md#network-volumes) cannot use imported backend fleets.
 
 ??? info "Required role"
     The user creating or updating an export must have the project admin role on both the exporter project and

src/dstack/_internal/server/background/pipeline_tasks/jobs_submitted.py

@@ -663,7 +663,9 @@ async def _apply_assignment_result(
                     await _reset_job_lock_for_retry(session=session, item=item)
                     return
                 fleet_spec = get_fleet_spec(fleet_model)
-                if not can_create_new_cloud_instance_in_fleet(fleet_model, fleet_spec):
+                if not can_create_new_cloud_instance_in_fleet(
+                    fleet_model, fleet_spec, job_project_id=context.project.id
+                ):
                     await _terminate_submitted_job(
                         session=session,
                         job_model=job_model,

src/dstack/_internal/server/services/exports.py

@@ -34,7 +34,7 @@
     ProjectRole,
     UserModel,
 )
-from dstack._internal.server.services.fleets import get_fleet_spec, list_project_fleet_models
+from dstack._internal.server.services.fleets import list_project_fleet_models
 from dstack._internal.server.services.gateways import list_project_gateway_models
 from dstack._internal.server.services.locking import get_locker, string_to_lock_id
 from dstack._internal.server.services.projects import (
@@ -387,13 +387,6 @@ async def add_exported_fleets(
         raise ResourceNotExistsError(
             f"Fleets {missing} not found in project {export.project.name!r}"
         )
-    cloud_fleet_names = [
-        f.name for f in fleets if get_fleet_spec(f).configuration.ssh_config is None
-    ]
-    if cloud_fleet_names:
-        raise ServerClientError(
-            f"Fleets {cloud_fleet_names} are cloud fleets. Can only export SSH fleets"
-        )
     for fleet in fleets:
         export.exported_fleets.append(ExportedFleetModel(fleet=fleet))
 

src/dstack/_internal/server/services/fleets.py

@@ -1006,9 +1006,13 @@ def get_fleet_master_instance_provisioning_data(
     return None
 
 
-def can_create_new_cloud_instance_in_fleet(fleet_model: FleetModel, fleet_spec: FleetSpec) -> bool:
+def can_create_new_cloud_instance_in_fleet(
+    fleet_model: FleetModel, fleet_spec: FleetSpec, job_project_id: uuid.UUID
+) -> bool:
     if fleet_spec.configuration.ssh_config is not None:
         return False
+    if fleet_model.project_id != job_project_id:
+        return False
     active_instances = [i for i in fleet_model.instances if i.status.is_active()]
     # nodes.max is a soft limit that can be exceeded when provisioning concurrently.
     # The fleet consolidation logic will remove redundant nodes eventually.
@@ -1021,8 +1025,10 @@ def can_create_new_cloud_instance_in_fleet(fleet_model: FleetModel, fleet_spec:
     return True
 
 
-def check_can_create_new_cloud_instance_in_fleet(fleet_model: FleetModel, fleet_spec: FleetSpec):
-    if not can_create_new_cloud_instance_in_fleet(fleet_model, fleet_spec):
+def check_can_create_new_cloud_instance_in_fleet(
+    fleet_model: FleetModel, fleet_spec: FleetSpec, job_project_id: uuid.UUID
+):
+    if not can_create_new_cloud_instance_in_fleet(fleet_model, fleet_spec, job_project_id):
         raise ValueError("Cannot fit new cloud instance into fleet")
 
 

src/dstack/_internal/server/services/runs/plan.py

@@ -392,6 +392,8 @@ async def find_optimal_fleet_with_offers(
     """
     if run_model is not None and run_model.fleet is not None:
         # Using the fleet that was already chosen by the master job
+        if volumes and run_model.fleet.project_id != project.id:
+            return run_model.fleet, [], []
         instance_offers = get_instance_offers_in_fleet(
             fleet_model=run_model.fleet,
             run_spec=run_spec,
@@ -412,6 +414,8 @@ async def find_optimal_fleet_with_offers(
     # First step: consider instance offers.
     candidates: list[_FleetCandidate] = []
     for fleet_model in fleet_models:
+        if volumes and fleet_model.project_id != project.id:
+            continue
         fleet_spec = get_fleet_spec(fleet_model)
         if (
             is_multinode_job(job)
@@ -420,7 +424,7 @@ async def find_optimal_fleet_with_offers(
             # Limit multinode runs to cluster fleets to guarantee best connectivity.
             continue
 
-        if not _run_can_fit_into_fleet(run_spec, fleet_model, fleet_spec):
+        if not _run_can_fit_into_fleet(project, run_spec, fleet_model, fleet_spec):
             logger.debug(
                 "Skipping fleet %s from consideration: run cannot fit into fleet",
                 fleet_model.name,
@@ -637,6 +641,8 @@ async def get_targeted_instance_offers(
         instance_ids=instance_ids,
         lock_instances=lock_instances,
     )
+    if volumes:
+        instances = [i for i in instances if i.project_id == project.id]
     return select_targeted_instance_offers(
         instances=instances,
         run_spec=run_spec,
@@ -721,7 +727,7 @@ def _get_required_instance_offers(run_spec: RunSpec, job: Job) -> int:
 
 
 def _run_can_fit_into_fleet(
-    run_spec: RunSpec, fleet_model: FleetModel, fleet_spec: FleetSpec
+    project: ProjectModel, run_spec: RunSpec, fleet_model: FleetModel, fleet_spec: FleetSpec
 ) -> bool:
     """
     Returns `False` if the run cannot fit into fleet for sure.
@@ -731,9 +737,19 @@ def _run_can_fit_into_fleet(
     or more than `nodes.max` instances being provisioned
     and eventually removed by the fleet consolidation logic.
     """
+    nodes_required_num = get_nodes_required_num(run_spec)
+    if fleet_spec.configuration.ssh_config is not None or fleet_model.project_id != project.id:
+        # SSH fleets and imported cloud fleets cannot provision new instances for the run,
+        # so only idle blocks of existing instances count.
+        # Currently assume that each idle block can run a job.
+        # TODO: Take resources / eligible offers into account.
+        total_idle_blocks = 0
+        for instance in fleet_model.instances:
+            if instance.status.is_available() and instance.total_blocks is not None:
+                total_idle_blocks += instance.total_blocks - instance.busy_blocks
+        return total_idle_blocks >= nodes_required_num
     # No check for cloud fleets with blocks > 1 since we don't know
     # how many jobs such fleets can accommodate.
-    nodes_required_num = get_nodes_required_num(run_spec)
     if (
         fleet_spec.configuration.nodes is not None
         and fleet_spec.configuration.blocks == 1
@@ -743,15 +759,6 @@ def _run_can_fit_into_fleet(
         fleet_available_capacity = fleet_spec.configuration.nodes.max - len(occupied_instances)
         if fleet_available_capacity < nodes_required_num:
             return False
-    elif fleet_spec.configuration.ssh_config is not None:
-        # Currently assume that each idle block can run a job.
-        # TODO: Take resources / eligible offers into account.
-        total_idle_blocks = 0
-        for instance in fleet_model.instances:
-            total_blocks = instance.total_blocks or 1
-            total_idle_blocks += total_blocks - instance.busy_blocks
-        if total_idle_blocks < nodes_required_num:
-            return False
     return True
 
 
@@ -776,7 +783,9 @@ async def _get_backend_offers_in_fleet(
     if fleet_spec is None:
         fleet_spec = get_fleet_spec(fleet_model)
     try:
-        check_can_create_new_cloud_instance_in_fleet(fleet_model, fleet_spec)
+        check_can_create_new_cloud_instance_in_fleet(
+            fleet_model, fleet_spec, job_project_id=project.id
+        )
         profile, requirements = get_run_profile_and_requirements_in_fleet(
             job=job,
             run_spec=run_spec,