Respect fleet specs when provisioning new instance on run apply

Author: r4victor

src/dstack/_internal/server/background/tasks/process_submitted_jobs.py

@@ -53,6 +53,7 @@
 from dstack._internal.server.services.backends import get_project_backend_by_type_or_error
 from dstack._internal.server.services.fleets import (
     fleet_model_to_fleet,
+    get_fleet_requirements,
 )
 from dstack._internal.server.services.instances import (
     filter_pool_instances,
@@ -71,6 +72,10 @@
 from dstack._internal.server.services.locking import get_locker
 from dstack._internal.server.services.logging import fmt
 from dstack._internal.server.services.offers import get_offers_by_requirements
+from dstack._internal.server.services.requirements.combine import (
+    combine_fleet_and_run_profiles,
+    combine_fleet_and_run_requirements,
+)
 from dstack._internal.server.services.runs import (
     check_run_spec_requires_instance_mounts,
     run_model_to_run,
@@ -646,6 +651,8 @@ async def _run_job_on_new_instance(
 ) -> Optional[Tuple[JobProvisioningData, InstanceOfferWithAvailability]]:
     if volumes is None:
         volumes = []
+    profile = run.run_spec.merged_profile
+    requirements = job.job_spec.requirements
     fleet = None
     if fleet_model is not None:
         fleet = fleet_model_to_fleet(fleet_model)
@@ -654,13 +661,26 @@ async def _run_job_on_new_instance(
                 "%s: cannot fit new instance into fleet %s", fmt(job_model), fleet_model.name
             )
             return None
+        profile = combine_fleet_and_run_profiles(fleet.spec.merged_profile, profile)
+        if profile is None:
+            logger.debug("%s: cannot combine fleet %s profile", fmt(job_model), fleet_model.name)
+            return None
+        fleet_requirements = get_fleet_requirements(fleet.spec)
+        requirements = combine_fleet_and_run_requirements(fleet_requirements, requirements)
+        if requirements is None:
+            logger.debug(
+                "%s: cannot combine fleet %s requirements", fmt(job_model), fleet_model.name
+            )
+            return None
+        # TODO: Respect fleet provisioning properties such as tags
+
     multinode = job.job_spec.jobs_per_replica > 1 or (
         fleet is not None and fleet.spec.configuration.placement == InstanceGroupPlacement.CLUSTER
     )
     offers = await get_offers_by_requirements(
         project=project,
-        profile=run.run_spec.merged_profile,
-        requirements=job.job_spec.requirements,
+        profile=profile,
+        requirements=requirements,
         exclude_not_available=True,
         multinode=multinode,
         master_job_provisioning_data=master_job_provisioning_data,

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

@@ -279,7 +279,7 @@ async def get_plan(
         offers_with_backends = await get_create_instance_offers(
             project=project,
             profile=effective_spec.merged_profile,
-            requirements=_get_fleet_requirements(effective_spec),
+            requirements=get_fleet_requirements(effective_spec),
             fleet_spec=effective_spec,
             blocks=effective_spec.configuration.blocks,
         )
@@ -458,7 +458,7 @@ async def create_fleet_instance_model(
     instance_num: int,
 ) -> InstanceModel:
     profile = spec.merged_profile
-    requirements = _get_fleet_requirements(spec)
+    requirements = get_fleet_requirements(spec)
     instance_model = await instances_services.create_instance_model(
         session=session,
         project=project,
@@ -644,6 +644,17 @@ def is_fleet_empty(fleet_model: FleetModel) -> bool:
     return len(active_instances) == 0
 
 
+def get_fleet_requirements(fleet_spec: FleetSpec) -> Requirements:
+    profile = fleet_spec.merged_profile
+    requirements = Requirements(
+        resources=fleet_spec.configuration.resources or ResourcesSpec(),
+        max_price=profile.max_price,
+        spot=get_policy_map(profile.spot_policy, default=SpotPolicy.ONDEMAND),
+        reservation=fleet_spec.configuration.reservation,
+    )
+    return requirements
+
+
 async def _create_fleet(
     session: AsyncSession,
     project: ProjectModel,
@@ -1004,17 +1015,6 @@ def _terminate_fleet_instances(fleet_model: FleetModel, instance_nums: Optional[
             instance.status = InstanceStatus.TERMINATING
 
 
-def _get_fleet_requirements(fleet_spec: FleetSpec) -> Requirements:
-    profile = fleet_spec.merged_profile
-    requirements = Requirements(
-        resources=fleet_spec.configuration.resources or ResourcesSpec(),
-        max_price=profile.max_price,
-        spot=get_policy_map(profile.spot_policy, default=SpotPolicy.ONDEMAND),
-        reservation=fleet_spec.configuration.reservation,
-    )
-    return requirements
-
-
 def _get_next_instance_num(instance_nums: set[int]) -> int:
     if not instance_nums:
         return 0