diff --git a/mkdocs/docs/concepts/exports.md b/mkdocs/docs/concepts/exports.md index d102835ce..72cc973d7 100644 --- a/mkdocs/docs/concepts/exports.md +++ b/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 diff --git a/src/dstack/_internal/server/background/pipeline_tasks/jobs_submitted.py b/src/dstack/_internal/server/background/pipeline_tasks/jobs_submitted.py index 58eaec1a6..ed0f3ecaa 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/jobs_submitted.py +++ b/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, diff --git a/src/dstack/_internal/server/services/exports.py b/src/dstack/_internal/server/services/exports.py index 4e90dfc2d..ecb136bd1 100644 --- a/src/dstack/_internal/server/services/exports.py +++ b/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)) diff --git a/src/dstack/_internal/server/services/fleets.py b/src/dstack/_internal/server/services/fleets.py index c959adc33..ca25bd99f 100644 --- a/src/dstack/_internal/server/services/fleets.py +++ b/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") diff --git a/src/dstack/_internal/server/services/runs/plan.py b/src/dstack/_internal/server/services/runs/plan.py index 28b6111ff..56590583a 100644 --- a/src/dstack/_internal/server/services/runs/plan.py +++ b/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, diff --git a/src/tests/_internal/server/background/pipeline_tasks/test_jobs_submitted.py b/src/tests/_internal/server/background/pipeline_tasks/test_jobs_submitted.py index e3360dd57..3d281939e 100644 --- a/src/tests/_internal/server/background/pipeline_tasks/test_jobs_submitted.py +++ b/src/tests/_internal/server/background/pipeline_tasks/test_jobs_submitted.py @@ -1771,8 +1771,9 @@ async def test_does_not_create_capacity_when_specific_instance_is_missing( res = await session.execute(select(InstanceModel)) assert res.scalars().all() == [] + @pytest.mark.parametrize("fleet_type", ["cloud", "ssh"]) async def test_assigns_job_to_imported_fleet( - self, test_db, session: AsyncSession, worker: JobSubmittedWorker + self, test_db, session: AsyncSession, worker: JobSubmittedWorker, fleet_type: str ): exporter_user = await create_user( session, name="exporter-user", global_role=GlobalRole.USER @@ -1787,11 +1788,11 @@ async def test_assigns_job_to_imported_fleet( session, name="importer-project", owner=importer_user ) repo = await create_repo(session=session, project_id=importer_project.id) - fleet = await create_fleet( - session=session, - project=exporter_project, - spec=get_fleet_spec(get_ssh_fleet_configuration()), - ) + if fleet_type == "cloud": + fleet_spec = get_fleet_spec() + else: + fleet_spec = get_fleet_spec(get_ssh_fleet_configuration()) + fleet = await create_fleet(session=session, project=exporter_project, spec=fleet_spec) instance = await create_instance( session=session, project=exporter_project, @@ -1820,6 +1821,204 @@ async def test_assigns_job_to_imported_fleet( assert job.instance is not None and job.instance.id == instance.id assert job.fleet_id == fleet.id + @pytest.mark.parametrize( + ("imported_idle_instances", "expected_fleet"), + [ + pytest.param(1, "own", id="imported-fleet-too-small"), + pytest.param(2, "imported", id="imported-fleet-fits"), + ], + ) + async def test_selects_imported_cloud_fleet_only_if_idle_instances_fit_run( + self, + test_db, + session: AsyncSession, + worker: JobSubmittedWorker, + imported_idle_instances: int, + expected_fleet: str, + ): + user = await create_user(session=session) + exporter_project = await create_project( + session=session, name="exporter-project", owner=user + ) + importer_project = await create_project( + session=session, name="importer-project", owner=user + ) + repo = await create_repo(session=session, project_id=importer_project.id) + imported_fleet_spec = get_fleet_spec() + imported_fleet_spec.configuration.nodes = FleetNodesSpec(min=0, target=0, max=4) + imported_fleet = await create_fleet( + session=session, project=exporter_project, spec=imported_fleet_spec + ) + for instance_num in range(imported_idle_instances): + await create_instance( + session=session, + project=exporter_project, + fleet=imported_fleet, + status=InstanceStatus.IDLE, + instance_num=instance_num, + price=1, + ) + await create_export( + session=session, + exporter_project=exporter_project, + importer_projects=[importer_project], + exported_fleets=[imported_fleet], + ) + own_fleet_spec = get_fleet_spec() + own_fleet_spec.configuration.nodes = FleetNodesSpec(min=0, target=0, max=2) + own_fleet = await create_fleet( + session=session, project=importer_project, spec=own_fleet_spec + ) + run_spec = get_run_spec( + repo_id=repo.name, + configuration=ServiceConfiguration( + port=8080, commands=["echo"], replicas=Range[int](min=2, max=2) + ), + ) + run = await create_run( + session=session, project=importer_project, repo=repo, user=user, run_spec=run_spec + ) + job = await create_job(session=session, run=run, replica_num=0) + + offer = get_instance_offer_with_availability(backend=BackendType.AWS, price=2) + backend_mock = Mock() + with patch( + "dstack._internal.server.services.backends.cache.lease_project_backends", + mock_lease([backend_mock]), + ): + backend_mock.TYPE = BackendType.AWS + backend_mock.compute.return_value.get_offers.return_value = [offer] + + await _process_job(session=session, worker=worker, job_model=job) + + job = await _get_job(session, job.id) + assert job.status == JobStatus.SUBMITTED + assert job.instance_assigned + if expected_fleet == "own": + assert job.fleet_id == own_fleet.id + else: + assert job.fleet_id == imported_fleet.id + + async def test_does_not_provision_new_capacity_in_imported_cloud_fleet( + self, test_db, session: AsyncSession, worker: JobSubmittedWorker + ): + user = await create_user(session=session) + exporter_project = await create_project( + session=session, name="exporter-project", owner=user + ) + importer_project = await create_project( + session=session, name="importer-project", owner=user + ) + repo = await create_repo(session=session, project_id=importer_project.id) + fleet_spec = get_fleet_spec() + fleet_spec.configuration.nodes = FleetNodesSpec(min=0, target=0, max=2) + fleet = await create_fleet(session=session, project=exporter_project, spec=fleet_spec) + await create_instance( + session=session, + project=exporter_project, + fleet=fleet, + status=InstanceStatus.BUSY, + ) + await create_export( + session=session, + exporter_project=exporter_project, + importer_projects=[importer_project], + exported_fleets=[fleet], + ) + run = await create_run(session=session, project=importer_project, repo=repo, user=user) + job = await create_job(session=session, run=run) + + offer = get_instance_offer_with_availability(backend=BackendType.AWS) + backend_mock = Mock() + with patch( + "dstack._internal.server.services.backends.cache.lease_project_backends", + mock_lease([backend_mock]), + ): + backend_mock.TYPE = BackendType.AWS + backend_mock.compute.return_value.get_offers.return_value = [offer] + + await _process_job(session=session, worker=worker, job_model=job) + + backend_mock.compute.return_value.get_offers.assert_not_called() + job = await _get_job(session, job.id) + assert job.status == JobStatus.TERMINATING + assert job.termination_reason == JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY + assert job.termination_reason_message == "Fleet is at capacity" + assert not job.instance_assigned + res = await session.execute( + select(InstanceModel).where( + InstanceModel.fleet_id == fleet.id, + InstanceModel.deleted == False, + ) + ) + assert len(res.scalars().all()) == 1 + + @pytest.mark.parametrize( + "run_pinned_to_fleet", + [ + pytest.param(False, id="new-run"), + # Volumes can be added by a rolling deployment after the run was placed + pytest.param(True, id="run-pinned-to-imported-fleet"), + ], + ) + async def test_does_not_assign_job_with_volumes_to_imported_fleet( + self, + test_db, + session: AsyncSession, + worker: JobSubmittedWorker, + run_pinned_to_fleet: bool, + ): + user = await create_user(session=session) + exporter_project = await create_project( + session=session, name="exporter-project", owner=user + ) + importer_project = await create_project( + session=session, name="importer-project", owner=user + ) + repo = await create_repo(session=session, project_id=importer_project.id) + volume = await create_volume( + session=session, + project=importer_project, + user=user, + status=VolumeStatus.ACTIVE, + volume_provisioning_data=get_volume_provisioning_data(), + backend=BackendType.AWS, + region="us-east-1", + ) + fleet = await create_fleet(session=session, project=exporter_project) + await create_instance( + session=session, + project=exporter_project, + fleet=fleet, + status=InstanceStatus.IDLE, + backend=BackendType.AWS, + region="us-east-1", + ) + await create_export( + session=session, + exporter_project=exporter_project, + importer_projects=[importer_project], + exported_fleets=[fleet], + ) + run_spec = get_run_spec(repo_id=repo.name) + run_spec.configuration.volumes = [VolumeMountPoint(name=volume.name, path="/volume")] + run = await create_run( + session=session, + project=importer_project, + repo=repo, + user=user, + run_spec=run_spec, + fleet=fleet if run_pinned_to_fleet else None, + ) + job = await create_job(session=session, run=run) + + await _process_job(session=session, worker=worker, job_model=job) + + job = await _get_job(session, job.id) + assert job.status == JobStatus.TERMINATING + assert job.termination_reason == JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY + assert not job.instance_assigned + async def test_assigns_job_to_specific_fleet( self, test_db, session: AsyncSession, worker: JobSubmittedWorker ): diff --git a/src/tests/_internal/server/routers/test_exports.py b/src/tests/_internal/server/routers/test_exports.py index 6bad45e9f..fea2c6229 100644 --- a/src/tests/_internal/server/routers/test_exports.py +++ b/src/tests/_internal/server/routers/test_exports.py @@ -111,9 +111,10 @@ async def test_creates_export( await create_fleet( session=session, project=project, - name="fleet1", + name="ssh-fleet", spec=get_fleet_spec(get_ssh_fleet_configuration()), ) + await create_fleet(session=session, project=project, name="cloud-fleet") backend = await create_backend(session=session, project_id=project.id) await create_gateway( session=session, project_id=project.id, backend_id=backend.id, name="gateway1" @@ -125,7 +126,7 @@ async def test_creates_export( json={ "name": "test-export", "importer_projects": ["ImporterProject"], - "exported_fleets": ["fleet1"], + "exported_fleets": ["ssh-fleet", "cloud-fleet"], "exported_gateways": ["gateway1"], }, ) @@ -135,8 +136,10 @@ async def test_creates_export( assert export_response["is_global"] == False assert len(export_response["imports"]) == 1 assert export_response["imports"][0]["project_name"] == "ImporterProject" - assert len(export_response["exported_fleets"]) == 1 - assert export_response["exported_fleets"][0]["name"] == "fleet1" + assert sorted(f["name"] for f in export_response["exported_fleets"]) == [ + "cloud-fleet", + "ssh-fleet", + ] assert len(export_response["exported_gateways"]) == 1 assert export_response["exported_gateways"][0]["name"] == "gateway1" @@ -255,14 +258,6 @@ async def test_creates_global_export(self, session: AsyncSession, client: AsyncC "Some gateways are listed for addition more than once", id="duplicate-gateway", ), - pytest.param( - { - "name": "test-export", - "exported_fleets": ["cloud-fleet"], - }, - "Fleets ['cloud-fleet'] are cloud fleets. Can only export SSH fleets", - id="cloud-fleet", - ), pytest.param( { "name": "test-export", @@ -314,7 +309,6 @@ async def test_rejects_invalid_export( name="exported-fleet", spec=get_fleet_spec(get_ssh_fleet_configuration()), ) - await create_fleet(session=session, project=project, name="cloud-fleet") backend = await create_backend(session=session, project_id=project.id) await create_gateway( session=session, @@ -737,14 +731,6 @@ async def test_can_add_same_entities_as_existing_deleted_ones( "Some gateways are listed for addition more than once", id="add-duplicate-gateway", ), - pytest.param( - { - "name": "test-export", - "add_exported_fleets": ["cloud-fleet"], - }, - "Fleets ['cloud-fleet'] are cloud fleets. Can only export SSH fleets", - id="add-cloud-fleet", - ), pytest.param( { "name": "test-export", @@ -933,7 +919,6 @@ async def test_rejects_invalid_update( exported_gateways=[exported_gateway], name="test-export", ) - await create_fleet(session=session, project=project, name="cloud-fleet") await create_fleet( session=session, project=project, diff --git a/src/tests/_internal/server/services/runs/test_plan.py b/src/tests/_internal/server/services/runs/test_plan.py index 3a75af257..04b07805f 100644 --- a/src/tests/_internal/server/services/runs/test_plan.py +++ b/src/tests/_internal/server/services/runs/test_plan.py @@ -46,6 +46,8 @@ get_job_provisioning_data, get_remote_connection_info, get_run_spec, + get_volume, + get_volume_configuration, mock_lease, ) @@ -524,6 +526,63 @@ async def test_returns_selected_instance_from_imported_fleet_reference( assert [instance for instance, _ in offers] == [selected] + @pytest.mark.asyncio + @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True) + async def test_skips_imported_instance_if_job_has_volumes( + self, test_db, session: AsyncSession + ) -> None: + user = await create_user(session=session) + project = await create_project(session=session, owner=user, name="importer-project") + exporter_project = await create_project( + session=session, owner=user, name="exporter-project" + ) + repo = await create_repo(session=session, project_id=project.id) + exported_fleet = await create_fleet( + session=session, project=exporter_project, name="exported-fleet" + ) + await create_instance( + session=session, + project=exporter_project, + fleet=exported_fleet, + instance_num=1, + backend=BackendType.AWS, + region="eu-west-1", + ) + await create_export( + session=session, + exporter_project=exporter_project, + importer_projects=[project], + exported_fleets=[exported_fleet], + ) + run_spec = get_run_spec( + repo_id=repo.name, + configuration=TaskConfiguration(image="debian", commands=["echo"]), + profile=Profile( + instances=[ + FleetInstanceSelector( + fleet=EntityReference.parse("exporter-project/exported-fleet"), + instance=1, + ) + ] + ), + ) + jobs = await get_jobs_from_run_spec(run_spec=run_spec, secrets={}, replica_num=0) + volume = get_volume( + project_name=project.name, + configuration=get_volume_configuration(backend=BackendType.AWS, region="eu-west-1"), + ) + + offers = await get_targeted_instance_offers( + session=session, + project=project, + run_spec=run_spec, + job=jobs[0], + volumes=[[volume]], + exclude_not_available=True, + ) + + assert offers == [] + @pytest.mark.asyncio @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True) async def test_returns_shared_block_offer_for_selected_instance( @@ -810,6 +869,50 @@ async def test_skips_backend_offers_when_instances_specified( class TestGetBackendOffersInFleet: + @pytest.mark.asyncio + @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True) + async def test_returns_no_offers_for_imported_fleet( + self, test_db, session: AsyncSession, monkeypatch: pytest.MonkeyPatch + ) -> None: + user = await create_user(session=session) + exporter_project = await create_project( + session=session, owner=user, name="exporter-project" + ) + importer_project = await create_project( + session=session, owner=user, name="importer-project" + ) + repo = await create_repo(session=session, project_id=importer_project.id) + fleet_spec = get_fleet_spec() + fleet_spec.configuration.nodes = FleetNodesSpec(min=0, target=0, max=2) + fleet = await create_fleet(session=session, project=exporter_project, spec=fleet_spec) + await create_export( + session=session, + exporter_project=exporter_project, + importer_projects=[importer_project], + exported_fleets=[fleet], + ) + run_spec = get_run_spec( + repo_id=repo.name, + configuration=TaskConfiguration(image="debian", commands=["echo"]), + ) + jobs = await get_jobs_from_run_spec(run_spec=run_spec, secrets={}, replica_num=0) + get_offers_by_requirements_mock = AsyncMock() + monkeypatch.setattr( + "dstack._internal.server.services.runs.plan.get_offers_by_requirements", + get_offers_by_requirements_mock, + ) + + offers = await _get_backend_offers_in_fleet( + project=importer_project, + fleet_model=fleet, + run_spec=run_spec, + job=jobs[0], + volumes=None, + ) + + assert offers == [] + get_offers_by_requirements_mock.assert_not_awaited() + @pytest.mark.asyncio @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True) async def test_keeps_unconstrained_offers_for_non_empty_cluster_fleet_without_elected_master(